diff --git a/CHANGELOG.md b/CHANGELOG.md
index ea1a8f346..d8c68b495 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,13 @@
 
 ## Unreleased
 
+- Embedded schedule audits default to 1,000 snapshots, preserving the first and
+  newest events. Compact firing receipts keep occurrence deduplication intact;
+  a namespace-scoped prune command removes rows after physical schedule purge.
+  Run the occurrence migration and stop old schedule readers before enabling
+  compaction. Explicitly disable compaction during a rolling upgrade as described
+  in `docs/schedule-audit-retention.md`.
+
 - Add a namespace-scoped open-run version-usage query with legacy/current/unknown
   counts and byte-exact keyset pages. Missing markers and unavailable history
   remain unknown when assessing legacy branch retirement.
diff --git a/docs/schedule-audit-retention.md b/docs/schedule-audit-retention.md
new file mode 100644
index 000000000..54fc3fd04
--- /dev/null
+++ b/docs/schedule-audit-retention.md
@@ -0,0 +1,66 @@
+# Schedule audit retention
+
+Embedded schedules keep at most 1,000 audit snapshots by default. The first
+retained event (normally `ScheduleCreated`) and the newest events remain in
+their original form, with their original sequence numbers. New events continue
+after the latest retained sequence. Set `workflows.v2.schedules.audit_history_limit`
+or `DW_V2_SCHEDULE_AUDIT_HISTORY_LIMIT` to an integer of at least 2 to change the
+cap. This behavior requires the schedule occurrence migration and updated
+scheduler code; it is source work pending publication.
+
+Every audit append compacts at most 100 old snapshots inside the schedule's
+write transaction. A newly created schedule stays within its configured cap.
+A legacy schedule already above the cap approaches it in bounded passes.
+To compact an idle legacy schedule, repeat:
+
+```sh
+php artisan workflow:v2:schedule-audit-prune --namespace=orders --schedule=daily-invoices
+```
+
+The JSON response reports deleted and physical retained counts, including rows
+hidden by application query filters. Stop when the retained count reaches the
+configured limit. An invalid limit refuses compaction.
+
+## Firing receipts and upgrades
+
+An audit snapshot is not the authority for whether an occurrence may start
+again. Compact occurrence receipts preserve that decision after snapshots are
+removed. They contain only the internal schedule ID, namespace and canonical
+UTC occurrence time. They remain until the schedule is physically purged;
+the snapshot cap does not bound this correctness data. Old trigger metadata
+is copied before its audit row is deleted. Malformed occurrence metadata refuses
+the transaction and retains the original audit evidence.
+
+Stop old scheduler and schedule lifecycle writer processes before enabling
+compaction. Older packages read
+audit history for deduplication and cannot read the new compact receipts. For
+a staged upgrade, explicitly set `audit_history_limit` to PHP `null` in all
+new processes until every old scheduler and lifecycle writer has stopped. This temporarily
+disables snapshot compaction. Then enable the configured cap. Setting this
+value to zero or an invalid string does not disable retention; it is an error.
+The migration creates the receipt table without scanning all existing history.
+A schema rollback refuses to discard populated receipt storage.
+
+## Purged schedules
+
+Deleting a schedule through the API marks it deleted and keeps its bounded
+audit history and firing receipts. Once the schedule row is physically removed,
+run the following command in that namespace:
+
+```sh
+php artisan workflow:v2:schedule-audit-prune --namespace=orders --limit=500
+```
+
+Each pass deletes at most the selected number of orphan audit rows and receipts
+combined. Repeat until both deleted counts are zero. A remaining schedule row,
+including a soft deleted schedule, protects its rows. Namespace comparisons are
+exact; the command never sweeps other namespaces. Omission selects the configured
+namespace, or `default` when none is configured. Use `--null-namespace` instead
+of `--namespace` only for legacy rows whose namespace is SQL `NULL`.
+
+Custom schedule, audit and occurrence models use the configured workflow storage
+connection and table prefix. Applications which replace tables must also provide
+the corresponding migration and configure `workflows.v2.schedule_occurrence_model`.
+Application query filters cannot hide firing receipts or make a physically
+existing schedule eligible for orphan cleanup. The maintenance paths enforce
+namespace and internal owner identity explicitly.
diff --git a/src/Commands/V2ScheduleAuditPruneCommand.php b/src/Commands/V2ScheduleAuditPruneCommand.php
new file mode 100644
index 000000000..24b3a8fcb
--- /dev/null
+++ b/src/Commands/V2ScheduleAuditPruneCommand.php
@@ -0,0 +1,70 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Workflow\Commands;
+
+use Illuminate\Console\Command;
+use LogicException;
+use Workflow\V2\Models\WorkflowSchedule;
+use Workflow\V2\Support\ConfiguredV2Models;
+use Workflow\V2\Support\NamespaceIdentity;
+use Workflow\V2\Support\ScheduleAuditRetention;
+
+class V2ScheduleAuditPruneCommand extends Command
+{
+    protected $signature = 'workflow:v2:schedule-audit-prune
+        {--namespace= : Exact namespace; defaults to configured namespace or default}
+        {--null-namespace : Select only legacy NULL-namespace rows}
+        {--schedule= : Compact this existing schedule ID instead of cleaning orphaned rows}
+        {--limit=500 : Maximum orphan rows deleted in this pass (1 to 1000)}';
+
+    protected $description = 'Bound schedule audit snapshots or clean audit rows left by purged schedules';
+
+    public function handle(): int
+    {
+        $namespace = $this->option('namespace');
+        $nullNamespace = (bool) $this->option('null-namespace');
+        if ($nullNamespace && $namespace !== null) {
+            $this->error('--namespace and --null-namespace cannot be combined.');
+            return self::FAILURE;
+        }
+        $namespace = $nullNamespace ? null : ($namespace ?? config('workflows.v2.namespace') ?? 'default');
+        if ($namespace !== null && (! is_string($namespace) || strlen($namespace) > 255
+            || preg_match('/(*UCP)\S/u', $namespace) !== 1)) {
+            $this->error('The selected namespace must be a nonempty UTF-8 string within 255 bytes.');
+            return self::FAILURE;
+        }
+        $limit = $this->option('limit');
+        if (is_string($limit) && preg_match('/\A[1-9][0-9]*\z/D', $limit) === 1) {
+            $limit = filter_var($limit, FILTER_VALIDATE_INT);
+        }
+        if (! is_int($limit) || $limit < 1 || $limit > 1000) {
+            $this->error('--limit must be an integer between 1 and 1000.');
+            return self::FAILURE;
+        }
+
+        try {
+            $id = $this->option('schedule');
+            if ($id === null) {
+                $report = ScheduleAuditRetention::cleanup($namespace, $limit);
+            } else {
+                $schedule = NamespaceIdentity::constrain(
+                    ConfiguredV2Models::query('schedule_model', WorkflowSchedule::class)->withoutGlobalScopes(),
+                    $namespace,
+                )->where('schedule_id', $id)
+                    ->first();
+                if ($schedule === null || $schedule->schedule_id !== $id) {
+                    $this->error('The exact schedule ID does not exist in the selected namespace.');
+                    return self::FAILURE;
+                }
+                $report = ScheduleAuditRetention::pruneWithReport($schedule);
+            }
+            $this->line(json_encode($report, JSON_THROW_ON_ERROR));
+        } catch (LogicException $error) {
+            $this->error($error->getMessage());
+            return self::FAILURE;
+        }
+        return self::SUCCESS;
+    }
+}
diff --git a/src/Providers/WorkflowServiceProvider.php b/src/Providers/WorkflowServiceProvider.php
index 6f86df1d1..747e29c44 100644
--- a/src/Providers/WorkflowServiceProvider.php
+++ b/src/Providers/WorkflowServiceProvider.php
@@ -24,6 +24,7 @@
 use Workflow\Commands\V2ReplayConformanceCommand;
 use Workflow\Commands\V2ReplaySimulateCommand;
 use Workflow\Commands\V2ReplayVerifyCommand;
+use Workflow\Commands\V2ScheduleAuditPruneCommand;
 use Workflow\Commands\V2ScheduleTickCommand;
 use Workflow\Commands\V2UpgradeStatusCommand;
 use Workflow\Commands\WorkflowMakeCommand;
@@ -206,6 +207,7 @@ public function boot(): void
             V2ReplaySimulateCommand::class,
             V2ReplayVerifyCommand::class,
             V2ScheduleTickCommand::class,
+            V2ScheduleAuditPruneCommand::class,
             V2UpgradeStatusCommand::class,
         ]);
 
diff --git a/src/V2/Models/WorkflowScheduleHistoryEvent.php b/src/V2/Models/WorkflowScheduleHistoryEvent.php
index 41f07400e..774dc996a 100644
--- a/src/V2/Models/WorkflowScheduleHistoryEvent.php
+++ b/src/V2/Models/WorkflowScheduleHistoryEvent.php
@@ -15,6 +15,7 @@
 use Workflow\V2\Enums\HistoryEventType;
 use Workflow\V2\Support\ConfiguredV2Models;
 use Workflow\V2\Support\HistoryEventPayloadContract;
+use Workflow\V2\Support\ScheduleAuditRetention;
 
 class WorkflowScheduleHistoryEvent extends Model
 {
@@ -63,6 +64,28 @@ public static function record(
         array $payload = [],
     ): self {
         HistoryEventPayloadContract::assertKnownPayloadKeys($eventType, $payload);
+        $snapshot = self::snapshotPayload($schedule, $payload);
+
+        return ScheduleAuditRetention::locked(
+            $schedule,
+            static fn (WorkflowSchedule $locked): self => self::recordLocked($locked, $eventType, $snapshot),
+        );
+    }
+
+    public static function utcOccurrenceKey(DateTimeInterface $occurrence): string
+    {
+        return \Carbon\CarbonImmutable::instance($occurrence)
+            ->utc()
+            ->format('Y-m-d H:i:s.u');
+    }
+
+    private static function recordLocked(
+        WorkflowSchedule $schedule,
+        HistoryEventType $eventType,
+        array $payload,
+    ): self {
+        // Validate configuration before appending an event.
+        ScheduleAuditRetention::limit();
 
         $snapshot = self::snapshotPayload($schedule, $payload);
         $workflowInstanceId = self::stringValue($payload['workflow_instance_id'] ?? null);
@@ -73,48 +96,56 @@ public static function record(
         $occurrenceAtUtc = $occurrenceTime !== null
             ? self::utcOccurrenceKey(new DateTimeImmutable($occurrenceTime))
             : null;
+        $historyModel = ConfiguredV2Models::resolve('schedule_history_event_model', self::class);
+        $history = (new $historyModel())->setConnection($schedule->getConnectionName())
+            ->newQueryWithoutScopes();
 
         for ($attempt = 1; $attempt <= self::SEQUENCE_RETRY_LIMIT; $attempt++) {
-            $sequence = ((int) ConfiguredV2Models::query('schedule_history_event_model', self::class)
-                ->where('workflow_schedule_id', $schedule->id)
-                ->max('sequence')) + 1;
+            $sequenceQuery = (clone $history)->where('workflow_schedule_id', $schedule->id);
+            // A committed old writer can win after our first read. On retry,
+            // bypass a repeatable-read snapshot without leaving this transaction.
+            $lastSequence = $attempt === 1 ? $sequenceQuery->max('sequence')
+                : $sequenceQuery->orderByDesc('sequence')
+                    ->lockForUpdate()
+                    ->value('sequence');
+            $sequence = ((int) $lastSequence) + 1;
+            $attributes = [
+                'workflow_schedule_id' => $schedule->id,
+                'schedule_id' => $schedule->schedule_id,
+                'namespace' => $schedule->namespace,
+                'sequence' => $sequence,
+                'event_type' => $eventType->value,
+                'payload' => $snapshot,
+                'workflow_instance_id' => $workflowInstanceId,
+                'workflow_run_id' => $workflowRunId,
+                'occurrence_at_utc' => $occurrenceAtUtc,
+                'recorded_at' => now(),
+            ];
 
             try {
                 /** @var self $event */
-                $event = ConfiguredV2Models::query('schedule_history_event_model', self::class)->create([
-                    'workflow_schedule_id' => $schedule->id,
-                    'schedule_id' => $schedule->schedule_id,
-                    'namespace' => $schedule->namespace,
-                    'sequence' => $sequence,
-                    'event_type' => $eventType->value,
-                    'payload' => $snapshot,
-                    'workflow_instance_id' => $workflowInstanceId,
-                    'workflow_run_id' => $workflowRunId,
-                    'occurrence_at_utc' => $occurrenceAtUtc,
-                    'recorded_at' => now(),
-                ]);
-
-                return $event;
+                $event = $schedule->getConnection()
+                    ->transaction(static fn (): self => (clone $history)->create($attributes));
             } catch (UniqueConstraintViolationException $e) {
                 if ($attempt === self::SEQUENCE_RETRY_LIMIT) {
                     throw $e;
                 }
                 // Another writer claimed this sequence first. Re-read the max
                 // and try the next slot.
+                continue;
             }
+            if ($occurrenceAtUtc !== null) {
+                ScheduleAuditRetention::retainOccurrence($schedule, $occurrenceAtUtc);
+            }
+            ScheduleAuditRetention::pruneLocked($schedule);
+
+            return $event;
         }
 
         // Unreachable: the loop either returns or re-throws at the final attempt.
         throw new \LogicException('Schedule history sequence allocation exhausted retries.');
     }
 
-    public static function utcOccurrenceKey(DateTimeInterface $occurrence): string
-    {
-        return \Carbon\CarbonImmutable::instance($occurrence)
-            ->utc()
-            ->format('Y-m-d H:i:s.u');
-    }
-
     /**
      * @param  array<string, mixed>  $payload
      * @return array<string, mixed>
diff --git a/src/V2/Models/WorkflowScheduleOccurrence.php b/src/V2/Models/WorkflowScheduleOccurrence.php
new file mode 100644
index 000000000..3c3b6d17f
--- /dev/null
+++ b/src/V2/Models/WorkflowScheduleOccurrence.php
@@ -0,0 +1,26 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Workflow\V2\Models;
+
+use Illuminate\Database\Eloquent\Concerns\HasUlids;
+use Illuminate\Database\Eloquent\Model;
+use Workflow\Traits\ResolvesStorageConnection;
+
+/** Compact firing receipts outlive the bounded audit snapshots. */
+class WorkflowScheduleOccurrence extends Model
+{
+    use HasUlids;
+    use ResolvesStorageConnection;
+
+    public $incrementing = false;
+
+    public $timestamps = false;
+
+    protected $table = 'workflow_schedule_occurrences';
+
+    protected $guarded = [];
+
+    protected $keyType = 'string';
+}
diff --git a/src/V2/Support/ScheduleAuditRetention.php b/src/V2/Support/ScheduleAuditRetention.php
new file mode 100644
index 000000000..a4e504e05
--- /dev/null
+++ b/src/V2/Support/ScheduleAuditRetention.php
@@ -0,0 +1,243 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Workflow\V2\Support;
+
+use Closure;
+use DateTimeImmutable;
+use DateTimeZone;
+use Illuminate\Database\Eloquent\Builder;
+use InvalidArgumentException;
+use LogicException;
+use Workflow\V2\Enums\HistoryEventType;
+use Workflow\V2\Models\WorkflowSchedule;
+use Workflow\V2\Models\WorkflowScheduleHistoryEvent;
+use Workflow\V2\Models\WorkflowScheduleOccurrence;
+
+/** @internal Retains audit snapshots without forgetting an admitted occurrence. */
+final class ScheduleAuditRetention
+{
+    public const DEFAULT_LIMIT = 1000;
+
+    private const BATCH_SIZE = 100;
+
+    public static function limit(): ?int
+    {
+        $value = config('workflows.v2.schedules.audit_history_limit', self::DEFAULT_LIMIT);
+        if ($value === null) {
+            return null;
+        }
+        if (is_string($value) && preg_match('/\A[1-9][0-9]*\z/D', $value) === 1) {
+            $value = filter_var($value, FILTER_VALIDATE_INT);
+        }
+        if (! is_int($value) || $value < 2) {
+            throw new InvalidArgumentException(
+                'Schedule audit history limit must be an integer of at least 2, or null.'
+            );
+        }
+        return $value;
+    }
+
+    /**
+     * @template T
+     * @param Closure(WorkflowSchedule): T $operation
+     * @return T
+     */
+    public static function locked(WorkflowSchedule $schedule, Closure $operation): mixed
+    {
+        $connection = $schedule->getConnection();
+        $model = ConfiguredV2Models::resolve('schedule_model', WorkflowSchedule::class);
+        $query = (new $model())->setConnection($connection->getName())
+            ->newQueryWithoutScopes();
+
+        return SqliteTransaction::run(static function () use ($query, $schedule, $operation): mixed {
+            $locked = (clone $query)->whereKey($schedule->getKey())
+                ->lockForUpdate()
+                ->firstOrFail();
+            if ($locked->getKey() !== $schedule->getKey() || $locked->namespace !== $schedule->namespace
+                || $locked->schedule_id !== $schedule->schedule_id) {
+                throw new LogicException('The schedule identity changed before audit retention acquired its lock.');
+            }
+            return $operation($locked);
+        }, admission: static function () use ($query, $schedule): void {
+            $key = $query->getModel()
+                ->getKeyName();
+            (clone $query)->whereKey($schedule->getKey())
+                ->toBase()
+                ->update([
+                    $key => $schedule->getKey(),
+                ]);
+        }, connection: $connection);
+    }
+
+    public static function prune(WorkflowSchedule $schedule): int
+    {
+        return self::pruneWithReport($schedule)['audit_events_deleted'];
+    }
+
+    /**
+     * @return array{audit_events_deleted: int, retained_events: int, configured_limit: int|null}
+     */
+    public static function pruneWithReport(WorkflowSchedule $schedule): array
+    {
+        return self::locked($schedule, static function (WorkflowSchedule $locked): array {
+            $deleted = self::pruneLocked($locked);
+            return [
+                'audit_events_deleted' => $deleted,
+                'retained_events' => self::history($locked)->count(),
+                'configured_limit' => self::limit(),
+            ];
+        });
+    }
+
+    /**
+     * Caller holds the schedule's write lock for the entire audit append/prune.
+     */
+    public static function pruneLocked(WorkflowSchedule $schedule): int
+    {
+        $limit = self::limit();
+        if ($limit === null) {
+            return 0;
+        }
+        $history = self::history($schedule);
+        $first = (clone $history)->min('sequence');
+        $cutoff = (clone $history)->orderByDesc('sequence')
+            ->skip($limit - 2)
+            ->value('sequence');
+        if ($first === null || $cutoff === null) {
+            return 0;
+        }
+        $events = (clone $history)->where('sequence', '>', $first)
+            ->where('sequence', '<', $cutoff)
+            ->orderBy('sequence')
+            ->limit(self::BATCH_SIZE)->toBase()
+            ->get(['id', 'event_type', 'occurrence_at_utc']);
+        foreach ($events as $event) {
+            if ($event->event_type !== HistoryEventType::ScheduleTriggered->value) {
+                continue;
+            }
+            $key = $event->occurrence_at_utc;
+            if ($key === null) {
+                // Extract metadata in SQL; do not load or decode the full
+                // schedule/action snapshot or invoke an application cast.
+                $clock = (clone $history)->whereKey($event->id)
+                    ->toBase()
+                    ->value('payload->occurrence_time');
+                if ($clock === null) {
+                    continue; // A manual trigger has no scheduled occurrence.
+                }
+                if (! is_string($clock) || strlen($clock) > 64
+                    || preg_match(
+                        '/\A[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}(?:\.[0-9]{1,6})?(?:Z|[+-][0-9]{2}:[0-9]{2})\z/D',
+                        $clock
+                    ) !== 1) {
+                    throw new LogicException(
+                        'A legacy schedule occurrence cannot be established; its audit row was retained.'
+                    );
+                }
+                try {
+                    $occurrence = new DateTimeImmutable($clock);
+                    $errors = DateTimeImmutable::getLastErrors();
+                    if ($errors !== false && ($errors['warning_count'] !== 0 || $errors['error_count'] !== 0)) {
+                        throw new LogicException('The legacy occurrence clock is invalid.');
+                    }
+                    $key = WorkflowScheduleHistoryEvent::utcOccurrenceKey($occurrence);
+                } catch (\Exception $error) {
+                    throw new LogicException(
+                        'A legacy schedule occurrence cannot be established; its audit row was retained.',
+                        0,
+                        $error
+                    );
+                }
+            }
+            self::retainOccurrence($schedule, $key);
+        }
+        if ($events->isEmpty()) {
+            return 0;
+        }
+        return (clone $history)->whereKey($events->pluck('id')->all())
+            ->delete();
+    }
+
+    public static function retainOccurrence(WorkflowSchedule $schedule, string $key): void
+    {
+        $clock = DateTimeImmutable::createFromFormat('!Y-m-d H:i:s.u', $key, new DateTimeZone('UTC'));
+        if ($clock === false || $clock->format('Y-m-d H:i:s.u') !== $key) {
+            throw new LogicException('A noncanonical schedule occurrence cannot replace retained audit evidence.');
+        }
+        $model = ConfiguredV2Models::resolve('schedule_occurrence_model', WorkflowScheduleOccurrence::class);
+        $receipt = (new $model())->setConnection($schedule->getConnectionName())
+            ->newQueryWithoutScopes()
+            ->firstOrCreate([
+                'workflow_schedule_id' => $schedule->getKey(),
+                'occurrence_at_utc' => $key,
+            ], [
+                'namespace' => $schedule->namespace,
+            ]);
+        if ($receipt->namespace !== $schedule->namespace) {
+            throw new LogicException('The schedule occurrence receipt has conflicting namespace ownership.');
+        }
+    }
+
+    /**
+     * Delete only rows whose internal schedule owner no longer exists.
+     */
+    public static function cleanup(?string $namespace, int $limit = 500): array
+    {
+        if ($limit < 1 || $limit > 1000) {
+            throw new InvalidArgumentException('Schedule audit cleanup limit must be between 1 and 1000.');
+        }
+        $schedule = ConfiguredV2Models::query('schedule_model', WorkflowSchedule::class)->getModel();
+        $connection = $schedule->getConnection();
+
+        return SqliteTransaction::run(static function () use ($namespace, $limit, $schedule, $connection): array {
+            $report = [
+                'audit_events_deleted' => 0,
+                'occurrences_deleted' => 0,
+            ];
+            $remaining = $limit;
+            foreach ([
+                ['schedule_history_event_model', WorkflowScheduleHistoryEvent::class, 'audit_events_deleted'],
+                ['schedule_occurrence_model', WorkflowScheduleOccurrence::class, 'occurrences_deleted'],
+            ] as [$config, $default, $field]) {
+                if ($remaining === 0) {
+                    break;
+                }
+                $model = ConfiguredV2Models::resolve($config, $default);
+                $query = NamespaceIdentity::constrain(
+                    (new $model())->setConnection($connection->getName())
+                        ->newQueryWithoutScopes(),
+                    $namespace,
+                );
+                $owners = $connection->table($schedule->getTable())
+                    ->selectRaw('1');
+                $owners->whereRaw(NamespaceIdentity::columnsEqual(
+                    $owners,
+                    $schedule->qualifyColumn('id'),
+                    $query->getModel()
+                        ->qualifyColumn('workflow_schedule_id'),
+                ));
+                $query->whereNotExists($owners);
+                $ids = (clone $query)->orderBy('id')
+                    ->limit($remaining)
+                    ->pluck('id')
+                    ->all();
+                if ($ids !== []) {
+                    $report[$field] = (clone $query)->whereKey($ids)->delete();
+                    $remaining -= $report[$field];
+                }
+            }
+            $report['limit_reached'] = $remaining === 0;
+            return $report;
+        }, connection: $connection);
+    }
+
+    private static function history(WorkflowSchedule $schedule): Builder
+    {
+        $model = ConfiguredV2Models::resolve('schedule_history_event_model', WorkflowScheduleHistoryEvent::class);
+        return (new $model())->setConnection($schedule->getConnectionName())
+            ->newQueryWithoutScopes()
+            ->where('workflow_schedule_id', $schedule->getKey());
+    }
+}
diff --git a/src/V2/Support/ScheduleManager.php b/src/V2/Support/ScheduleManager.php
index d8f6591e2..884b11110 100644
--- a/src/V2/Support/ScheduleManager.php
+++ b/src/V2/Support/ScheduleManager.php
@@ -25,6 +25,7 @@
 use Workflow\V2\Models\WorkflowRun;
 use Workflow\V2\Models\WorkflowSchedule;
 use Workflow\V2\Models\WorkflowScheduleHistoryEvent;
+use Workflow\V2\Models\WorkflowScheduleOccurrence;
 use Workflow\V2\StartOptions;
 use Workflow\V2\WorkflowStub;
 
@@ -976,7 +977,15 @@ private static function hasFiredOccurrence(
         DateTimeInterface $occurrenceTime,
     ): bool {
         $key = WorkflowScheduleHistoryEvent::utcOccurrenceKey($occurrenceTime);
+        if (ConfiguredV2Models::query('schedule_occurrence_model', WorkflowScheduleOccurrence::class)
+            ->withoutGlobalScopes()
+            ->where('workflow_schedule_id', $schedule->id)
+            ->where('occurrence_at_utc', $key)
+            ->exists()) {
+            return true;
+        }
         $history = ConfiguredV2Models::query('schedule_history_event_model', WorkflowScheduleHistoryEvent::class)
+            ->withoutGlobalScopes()
             ->where('workflow_schedule_id', $schedule->id)
             ->where('event_type', HistoryEventType::ScheduleTriggered->value);
 
diff --git a/src/config/workflows.php b/src/config/workflows.php
index ad566813d..ebe333ef6 100644
--- a/src/config/workflows.php
+++ b/src/config/workflows.php
@@ -91,6 +91,7 @@
         'run_lineage_entry_model' => Workflow\V2\Models\WorkflowRunLineageEntry::class,
         'schedule_model' => Workflow\V2\Models\WorkflowSchedule::class,
         'schedule_history_event_model' => Workflow\V2\Models\WorkflowScheduleHistoryEvent::class,
+        'schedule_occurrence_model' => Workflow\V2\Models\WorkflowScheduleOccurrence::class,
         // One-time upgrade conversion of pre-UTC schedule timestamp columns.
         // Set this to the PHP timezone used when the existing rows were written
         // if it differs from the application's timezone during migration.
@@ -412,6 +413,14 @@
         ],
 
         'schedules' => [
+            // Keep the first audit event and the newest remaining events.
+            // Explicit null disables compaction during a rolling upgrade;
+            // enable it only after all old schedule readers have stopped.
+            'audit_history_limit' => Env::dw(
+                'DW_V2_SCHEDULE_AUDIT_HISTORY_LIMIT',
+                'WORKFLOW_V2_SCHEDULE_AUDIT_HISTORY_LIMIT',
+                Workflow\V2\Support\ScheduleAuditRetention::DEFAULT_LIMIT,
+            ),
             // Pass size for `workflow:v2:schedule-tick`. Each pass starts at
             // most this many due schedules, plus at most this many buffered
             // actions, shared round-robin across namespaces (oldest first).
diff --git a/src/migrations/2026_11_14_000200_create_workflow_schedule_occurrences.php b/src/migrations/2026_11_14_000200_create_workflow_schedule_occurrences.php
new file mode 100644
index 000000000..a1379a27b
--- /dev/null
+++ b/src/migrations/2026_11_14_000200_create_workflow_schedule_occurrences.php
@@ -0,0 +1,45 @@
+<?php
+
+declare(strict_types=1);
+
+use Illuminate\Database\Schema\Blueprint;
+use Illuminate\Support\Facades\DB;
+use Illuminate\Support\Facades\Schema;
+use Workflow\Support\WorkflowMigration;
+
+return new class() extends WorkflowMigration {
+    private const TABLE = 'workflow_schedule_occurrences';
+
+    public function up(): void
+    {
+        if (Schema::hasTable(self::TABLE)) {
+            return;
+        }
+
+        $this->withSchemaLockBudget(static function (): void {
+            Schema::create(self::TABLE, static function (Blueprint $table): void {
+                $table->string('id', 26)
+                    ->primary();
+                $table->string('workflow_schedule_id', 26);
+                $table->string('namespace')
+                    ->nullable()
+                    ->index();
+                $table->string('occurrence_at_utc', 26);
+                $table->unique(['workflow_schedule_id', 'occurrence_at_utc'], 'wf_schedule_occurrence_unique');
+            });
+        });
+    }
+
+    public function down(): void
+    {
+        if (! Schema::hasTable(self::TABLE)) {
+            return;
+        }
+        // Removing receipts after audit compaction could admit a duplicate
+        // historical backfill. A rollback needs a restored pre-compaction DB.
+        if (DB::table(self::TABLE)->exists()) {
+            throw new LogicException('Schedule occurrence receipts must be preserved after audit compaction.');
+        }
+        Schema::drop(self::TABLE);
+    }
+};
diff --git a/tests/Feature/V2/V2ScheduleAuditRetentionTest.php b/tests/Feature/V2/V2ScheduleAuditRetentionTest.php
new file mode 100644
index 000000000..cc7601448
--- /dev/null
+++ b/tests/Feature/V2/V2ScheduleAuditRetentionTest.php
@@ -0,0 +1,700 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Tests\Feature\V2;
+
+use Illuminate\Database\Eloquent\Builder;
+use Illuminate\Support\Carbon;
+use Illuminate\Support\Facades\Artisan;
+use Illuminate\Support\Facades\Schema;
+use InvalidArgumentException;
+use LogicException;
+use PHPUnit\Framework\Attributes\DataProvider;
+use Tests\Fixtures\V2\TestScheduledWorkflow;
+use Tests\TestCase;
+use Workflow\V2\Contracts\ScheduleWorkflowStarter;
+use Workflow\V2\Enums\HistoryEventType;
+use Workflow\V2\Enums\ScheduleOverlapPolicy;
+use Workflow\V2\Models\WorkflowSchedule;
+use Workflow\V2\Models\WorkflowScheduleHistoryEvent;
+use Workflow\V2\Models\WorkflowScheduleOccurrence;
+use Workflow\V2\Support\ScheduleAuditRetention;
+use Workflow\V2\Support\ScheduleManager;
+use Workflow\V2\Support\ScheduleStartResult;
+
+final class V2ScheduleAuditRetentionTest extends TestCase
+{
+    public function testRepeatedUpdatesKeepCreationAndNewestEventsWithoutReusingSequences(): void
+    {
+        config()->set('workflows.v2.schedules.audit_history_limit', 4);
+        $schedule = $this->schedule('bounded');
+        $creation = $schedule->historyEvents()
+            ->firstOrFail()
+            ->toArray();
+        for ($i = 1; $i <= 20; ++$i) {
+            ScheduleManager::update($schedule->fresh(), notes: 'revision ' . $i);
+            $this->assertLessThanOrEqual(4, $schedule->historyEvents()->count());
+        }
+        $events = $schedule->historyEvents()
+            ->orderBy('sequence')
+            ->get();
+        $this->assertSame([1, 19, 20, 21], $events->pluck('sequence')->all());
+        $this->assertSame($creation, $events->first()->toArray());
+        $this->assertSame('revision 20', $events->last()->payload['schedule']['note']);
+
+        config()
+            ->set('workflows.v2.schedules.audit_history_limit', 2);
+        ScheduleManager::update($schedule->fresh(), notes: 'revision 21');
+        $this->assertSame([1, 22], $schedule->historyEvents()->orderBy('sequence')->pluck('sequence')->all());
+    }
+
+    public function testExplicitlyDisabledCompactionSupportsAnUpgradeBeforeOldReadersStop(): void
+    {
+        config()->set('workflows.v2.schedules.audit_history_limit', null);
+        $schedule = $this->schedule('upgrade');
+        for ($i = 0; $i < 8; ++$i) {
+            ScheduleManager::update($schedule->fresh(), notes: 'revision ' . $i);
+        }
+        $this->assertSame(9, $schedule->historyEvents()->count());
+        config()
+            ->set('workflows.v2.schedules.audit_history_limit', 3);
+        $this->assertSame(6, ScheduleAuditRetention::prune($schedule));
+        $this->assertSame([1, 8, 9], $schedule->historyEvents()->orderBy('sequence')->pluck('sequence')->all());
+    }
+
+    public function testSnapshotStillRepresentsTheCallerStateWithoutChangingThePersistedOwner(): void
+    {
+        $schedule = $this->schedule('snapshot', 'orders');
+        $persisted = $schedule->note;
+        $schedule->note = 'caller snapshot';
+        $event = WorkflowScheduleHistoryEvent::record($schedule, HistoryEventType::ScheduleUpdated);
+        $this->assertSame('caller snapshot', $event->payload['schedule']['note']);
+        $this->assertSame($persisted, $schedule->fresh()->note);
+        $this->assertSame('orders', $event->namespace);
+    }
+
+    public function testAStaleOrForgedNamespaceCannotCompactAnotherOwner(): void
+    {
+        $schedule = $this->schedule('owner', 'orders');
+        $before = $schedule->historyEvents()
+            ->get()
+            ->toArray();
+        $schedule->namespace = 'other';
+        try {
+            ScheduleAuditRetention::prune($schedule);
+            $this->fail('A conflicting schedule namespace was admitted.');
+        } catch (LogicException $error) {
+            $this->assertStringContainsString('schedule identity changed', $error->getMessage());
+        }
+        $this->assertSame($before, $schedule->historyEvents()->get()->toArray());
+        $this->assertSame(0, WorkflowScheduleOccurrence::query()->count());
+    }
+
+    public function testPrunedTriggerStillPreventsASecondStartForItsOccurrence(): void
+    {
+        config()->set('workflows.v2.schedules.audit_history_limit', 3);
+        $starter = $this->createMock(ScheduleWorkflowStarter::class);
+        $starter->expects($this->once())
+            ->method('start')
+            ->willReturn(new ScheduleStartResult('first-run', null));
+        app()
+            ->instance(ScheduleWorkflowStarter::class, $starter);
+        Carbon::setTestNow(Carbon::parse('2026-11-14T00:59:00Z'));
+        try {
+            $schedule = $this->schedule('deduplication');
+            $due = $schedule->next_fire_at;
+            Carbon::setTestNow($due);
+            $this->assertSame('triggered', ScheduleManager::tick()[0]['outcome']);
+            for ($i = 0; $i < 6; ++$i) {
+                ScheduleManager::update($schedule->fresh(), notes: 'revision ' . $i);
+            }
+            $this->assertFalse(
+                $schedule->historyEvents()
+                    ->where('event_type', HistoryEventType::ScheduleTriggered->value)->exists()
+            );
+            $this->assertSame(
+                1,
+                WorkflowScheduleOccurrence::query()->where('workflow_schedule_id', $schedule->id)->count()
+            );
+            $answer = ScheduleManager::backfill($schedule->fresh(), $due, $due->copy()->addMinute());
+            $this->assertNull($answer[0]['instance_id']);
+            $this->assertArrayNotHasKey('error', $answer[0]);
+            $this->assertSame(1, (int) $schedule->fresh()->fires_count);
+            $this->assertSame('already_fired_occurrence', $schedule->fresh()->last_skip_reason);
+        } finally {
+            Carbon::setTestNow();
+        }
+    }
+
+    public function testLegacyTriggerMetadataMovesBeforeItsAuditRowIsDeleted(): void
+    {
+        config()->set('workflows.v2.schedules.audit_history_limit', null);
+        $schedule = $this->schedule('legacy');
+        $event = WorkflowScheduleHistoryEvent::record($schedule, HistoryEventType::ScheduleTriggered, [
+            'occurrence_time' => '2026-11-13T01:02:03.123456-05:00',
+        ]);
+        WorkflowScheduleOccurrence::query()->delete();
+        $event->forceFill([
+            'occurrence_at_utc' => null,
+        ])->save();
+        for ($i = 0; $i < 4; ++$i) {
+            ScheduleManager::update($schedule->fresh(), notes: 'revision ' . $i);
+        }
+        config()
+            ->set('workflows.v2.schedules.audit_history_limit', 3);
+        ScheduleAuditRetention::prune($schedule);
+        $this->assertModelMissing($event);
+        $this->assertSame(
+            '2026-11-13 06:02:03.123456',
+            WorkflowScheduleOccurrence::query()->firstOrFail()->occurrence_at_utc
+        );
+        $this->assertSame(3, $schedule->historyEvents()->count());
+    }
+
+    public function testAnUnusableLegacyOccurrenceRollsBackTheWholeCompaction(): void
+    {
+        config()->set('workflows.v2.schedules.audit_history_limit', null);
+        $schedule = $this->schedule('malformed');
+        WorkflowScheduleHistoryEvent::record($schedule, HistoryEventType::ScheduleTriggered, [
+            'occurrence_time' => '2026-11-13T01:02:03Z',
+        ]);
+        $bad = WorkflowScheduleHistoryEvent::record($schedule, HistoryEventType::ScheduleTriggered, [
+            'occurrence_time' => '2026-11-13T02:02:03Z',
+        ]);
+        WorkflowScheduleOccurrence::query()->delete();
+        $bad->forceFill([
+            'occurrence_at_utc' => 'malformed',
+        ])->save();
+        for ($i = 0; $i < 3; ++$i) {
+            ScheduleManager::update($schedule->fresh(), notes: 'revision ' . $i);
+        }
+        $before = $schedule->historyEvents()
+            ->orderBy('sequence')
+            ->get()
+            ->toArray();
+        config()
+            ->set('workflows.v2.schedules.audit_history_limit', 2);
+        try {
+            ScheduleAuditRetention::prune($schedule);
+            $this->fail('Unusable occurrence metadata was discarded.');
+        } catch (LogicException $error) {
+            $this->assertStringContainsString('noncanonical schedule occurrence', $error->getMessage());
+        }
+        $this->assertSame($before, $schedule->historyEvents()->orderBy('sequence')->get()->toArray());
+        $this->assertSame(0, WorkflowScheduleOccurrence::query()->count());
+    }
+
+    public function testCleanupKeepsSoftDeletedAndOtherNamespaceSchedules(): void
+    {
+        $kept = $this->schedule('kept', 'orders');
+        ScheduleManager::delete($kept);
+        $purged = $this->schedule('purged', 'orders');
+        WorkflowScheduleHistoryEvent::record($purged, HistoryEventType::ScheduleTriggered, [
+            'occurrence_time' => '2026-11-13T01:02:03Z',
+        ]);
+        $other = $this->schedule('other', 'Orders');
+        $other->delete();
+        $purged->delete();
+        $before = $kept->historyEvents()
+            ->count();
+        $first = ScheduleAuditRetention::cleanup('orders', 1);
+        $this->assertSame(1, $first['audit_events_deleted']);
+        $this->assertTrue($first['limit_reached']);
+        $rest = ScheduleAuditRetention::cleanup('orders');
+        $this->assertSame(1, $rest['audit_events_deleted']);
+        $this->assertSame(1, $rest['occurrences_deleted']);
+        $this->assertSame($before, $kept->historyEvents()->count());
+        $this->assertSame(1, $other->historyEvents()->count());
+    }
+
+    public function testApplicationFiltersCannotMakeAPhysicalOwnerAnOrphan(): void
+    {
+        $schedule = $this->schedule('filtered-owner', 'orders');
+        WorkflowScheduleHistoryEvent::record($schedule, HistoryEventType::ScheduleTriggered, [
+            'occurrence_time' => '2026-11-13T01:00:00Z',
+        ]);
+        config()
+            ->set('workflows.v2.schedule_model', HiddenScheduleForAuditRetention::class);
+        $report = ScheduleAuditRetention::cleanup('orders');
+        $this->assertSame(0, $report['audit_events_deleted']);
+        $this->assertSame(0, $report['occurrences_deleted']);
+        $this->assertSame(2, $schedule->historyEvents()->count());
+        $this->assertSame(1, WorkflowScheduleOccurrence::query()->count());
+    }
+
+    public function testApplicationFiltersCannotHideACompactedFiringReceipt(): void
+    {
+        config()->set('workflows.v2.schedules.audit_history_limit', 2);
+        $schedule = $this->schedule('filtered-receipt');
+        WorkflowScheduleHistoryEvent::record($schedule, HistoryEventType::ScheduleTriggered, [
+            'occurrence_time' => '2026-11-13T01:00:00Z',
+        ]);
+        WorkflowScheduleHistoryEvent::record($schedule, HistoryEventType::ScheduleUpdated);
+        $this->assertFalse(
+            $schedule->historyEvents()
+                ->where('event_type', HistoryEventType::ScheduleTriggered->value)->exists()
+        );
+        config()
+            ->set('workflows.v2.schedule_occurrence_model', HiddenOccurrenceForAuditRetention::class);
+        $starter = $this->createStub(ScheduleWorkflowStarter::class);
+        $starter->method('start')
+            ->willReturn(new ScheduleStartResult('unexpected-duplicate', null));
+        app()
+            ->instance(ScheduleWorkflowStarter::class, $starter);
+        $from = Carbon::parse('2026-11-13T01:00:00Z');
+        $answer = ScheduleManager::backfill($schedule, $from, $from->copy()->addMinute());
+        $this->assertNull($answer[0]['instance_id']);
+        $this->assertArrayNotHasKey('error', $answer[0]);
+        $this->assertSame('already_fired_occurrence', $schedule->fresh()->last_skip_reason);
+    }
+
+    public function testApplicationFiltersCannotHideAnUncompactedLegacyTrigger(): void
+    {
+        config()->set('workflows.v2.schedules.audit_history_limit', null);
+        $schedule = $this->schedule('filtered-legacy');
+        WorkflowScheduleHistoryEvent::record($schedule, HistoryEventType::ScheduleTriggered, [
+            'occurrence_time' => '2026-11-13T01:00:00Z',
+        ]);
+        WorkflowScheduleOccurrence::query()->delete();
+        config()
+            ->set('workflows.v2.schedule_history_event_model', HiddenScheduleHistoryForAuditRetention::class);
+        $starter = $this->createStub(ScheduleWorkflowStarter::class);
+        $starter->method('start')
+            ->willReturn(new ScheduleStartResult('unexpected-legacy-duplicate', null));
+        app()
+            ->instance(ScheduleWorkflowStarter::class, $starter);
+        $from = Carbon::parse('2026-11-13T01:00:00Z');
+        $answer = ScheduleManager::backfill($schedule, $from, $from->copy()->addMinute());
+        $this->assertNull($answer[0]['instance_id']);
+        $this->assertArrayNotHasKey('error', $answer[0]);
+        $this->assertSame('already_fired_occurrence', $schedule->fresh()->last_skip_reason);
+    }
+
+    public function testARecreatedScheduleDoesNotInheritThePurgedOwnersFiringReceipt(): void
+    {
+        $old = $this->schedule('recreated', 'orders');
+        WorkflowScheduleHistoryEvent::record($old, HistoryEventType::ScheduleTriggered, [
+            'occurrence_time' => '2026-11-13T01:00:00Z',
+        ]);
+        $old->delete();
+        $new = $this->schedule('recreated', 'orders');
+        $this->assertNotSame($old->id, $new->id);
+        $starter = $this->createMock(ScheduleWorkflowStarter::class);
+        $starter->expects($this->once())
+            ->method('start')
+            ->willReturn(new ScheduleStartResult('new-owner', null));
+        app()
+            ->instance(ScheduleWorkflowStarter::class, $starter);
+        $from = Carbon::parse('2026-11-13T01:00:00Z');
+        $answer = ScheduleManager::backfill($new, $from, $from->copy()->addMinute());
+        $this->assertSame('new-owner', $answer[0]['instance_id']);
+        $this->assertSame(1, (int) $new->fresh()->fires_count);
+        $this->assertSame(2, WorkflowScheduleOccurrence::query()->count());
+    }
+
+    public function testARefusedCompactionRollsBackTheNewAuditAppendAndItsFiringReceipt(): void
+    {
+        config()->set('workflows.v2.schedules.audit_history_limit', null);
+        $schedule = $this->schedule('append-rollback');
+        $bad = WorkflowScheduleHistoryEvent::record($schedule, HistoryEventType::ScheduleTriggered, [
+            'occurrence_time' => '2026-11-13T01:00:00Z',
+        ]);
+        $bad->forceFill([
+            'occurrence_at_utc' => 'malformed',
+        ])->save();
+        WorkflowScheduleOccurrence::query()->delete();
+        WorkflowScheduleHistoryEvent::record($schedule, HistoryEventType::ScheduleUpdated);
+        $before = $schedule->historyEvents()
+            ->orderBy('sequence')
+            ->get()
+            ->toArray();
+        config()
+            ->set('workflows.v2.schedules.audit_history_limit', 2);
+        try {
+            WorkflowScheduleHistoryEvent::record($schedule, HistoryEventType::ScheduleTriggered, [
+                'occurrence_time' => '2026-11-13T02:00:00Z',
+            ]);
+            $this->fail('An audit append escaped the refused compaction transaction.');
+        } catch (LogicException $error) {
+            $this->assertStringContainsString('noncanonical schedule occurrence', $error->getMessage());
+        }
+        $this->assertSame($before, $schedule->historyEvents()->orderBy('sequence')->get()->toArray());
+        $this->assertSame(0, WorkflowScheduleOccurrence::query()->count());
+        config()
+            ->set('workflows.v2.schedules.audit_history_limit', null);
+        $next = WorkflowScheduleHistoryEvent::record($schedule, HistoryEventType::ScheduleUpdated);
+        $this->assertSame(4, $next->sequence);
+    }
+
+    public function testLegacyBacklogIsCompactedInBoundedPasses(): void
+    {
+        config()->set('workflows.v2.schedules.audit_history_limit', null);
+        $schedule = $this->schedule('backlog');
+        for ($i = 0; $i < 205; ++$i) {
+            WorkflowScheduleHistoryEvent::record($schedule, HistoryEventType::ScheduleUpdated);
+        }
+        config()
+            ->set('workflows.v2.schedules.audit_history_limit', 3);
+        $this->assertSame(100, ScheduleAuditRetention::prune($schedule));
+        $this->assertSame(106, $schedule->historyEvents()->count());
+        $this->assertSame(100, ScheduleAuditRetention::prune($schedule));
+        $this->assertSame(3, ScheduleAuditRetention::prune($schedule));
+        $this->assertSame(0, ScheduleAuditRetention::prune($schedule));
+        $this->assertSame([1, 205, 206], $schedule->historyEvents()->orderBy('sequence')->pluck('sequence')->all());
+    }
+
+    public function testCleanupCommandUsesAnExactNamespaceAndHonorsItsLimit(): void
+    {
+        $target = $this->schedule('target', 'orders');
+        WorkflowScheduleHistoryEvent::record($target, HistoryEventType::ScheduleUpdated);
+        $other = $this->schedule('other-command', 'Orders');
+        $target->delete();
+        $other->delete();
+        $this->artisan('workflow:v2:schedule-audit-prune', [
+            '--namespace' => 'orders',
+            '--limit' => '1',
+        ])
+            ->assertExitCode(0)
+            ->run();
+        $this->assertSame(1, $target->historyEvents()->count());
+        $this->assertSame(1, $other->historyEvents()->count());
+        $this->artisan('workflow:v2:schedule-audit-prune', [
+            '--namespace' => 'orders',
+        ])
+            ->assertExitCode(0)
+            ->run();
+        $this->assertSame(0, $target->historyEvents()->count());
+        $this->assertSame(1, $other->historyEvents()->count());
+    }
+
+    public function testCommandCanCompactAnIdleScheduleAndSelectLegacyNullRowsSeparately(): void
+    {
+        config()->set('workflows.v2.schedules.audit_history_limit', null);
+        $schedule = $this->schedule('idle', 'orders');
+        for ($i = 0; $i < 5; ++$i) {
+            WorkflowScheduleHistoryEvent::record($schedule, HistoryEventType::ScheduleUpdated);
+        }
+        config()
+            ->set('workflows.v2.schedules.audit_history_limit', 2);
+        $this->artisan('workflow:v2:schedule-audit-prune', [
+            '--namespace' => 'orders',
+            '--schedule' => 'idle',
+        ])
+            ->assertExitCode(0)
+            ->run();
+        $this->assertSame([1, 6], $schedule->historyEvents()->orderBy('sequence')->pluck('sequence')->all());
+        $schedule->historyEvents()
+            ->update([
+                'namespace' => null,
+            ]);
+        $schedule->delete();
+        $this->artisan('workflow:v2:schedule-audit-prune', [
+            '--namespace' => 'orders',
+        ])
+            ->assertExitCode(0)
+            ->run();
+        $this->assertSame(2, $schedule->historyEvents()->count());
+        $this->artisan('workflow:v2:schedule-audit-prune', [
+            '--null-namespace' => true,
+        ])
+            ->assertExitCode(0)
+            ->run();
+        $this->assertSame(0, $schedule->historyEvents()->count());
+    }
+
+    public function testInvalidLimitsAndConflictingNamespaceOptionsRefuseDeletion(): void
+    {
+        $schedule = $this->schedule('invalid', 'orders');
+        $schedule->delete();
+        foreach (['0', '1001', 'invalid'] as $limit) {
+            $this->artisan('workflow:v2:schedule-audit-prune', [
+                '--namespace' => 'orders',
+                '--limit' => $limit,
+            ])
+                ->assertExitCode(1)
+                ->run();
+        }
+        $this->artisan('workflow:v2:schedule-audit-prune', [
+            '--namespace' => 'orders',
+            '--null-namespace' => true,
+        ])
+            ->assertExitCode(1)
+            ->run();
+        $this->assertSame(1, $schedule->historyEvents()->count());
+        config()
+            ->set('workflows.v2.schedules.audit_history_limit', 1);
+        $this->expectException(InvalidArgumentException::class);
+        ScheduleAuditRetention::limit();
+    }
+
+    public function testCommandReportsPhysicalRemainingRowsDespiteAnApplicationFilter(): void
+    {
+        config()->set('workflows.v2.schedules.audit_history_limit', null);
+        $schedule = $this->schedule('filtered-report', 'orders');
+        for ($i = 0; $i < 205; ++$i) {
+            WorkflowScheduleHistoryEvent::record($schedule, HistoryEventType::ScheduleUpdated);
+        }
+        config()
+            ->set('workflows.v2.schedule_history_event_model', HiddenScheduleHistoryForAuditRetention::class);
+        config()
+            ->set('workflows.v2.schedules.audit_history_limit', 3);
+        foreach ([[100, 106], [100, 6], [3, 3]] as [$deleted, $remaining]) {
+            $this->assertSame(0, Artisan::call('workflow:v2:schedule-audit-prune', [
+                '--namespace' => 'orders',
+                '--schedule' => 'filtered-report',
+            ]));
+            $report = json_decode(Artisan::output(), true, flags: JSON_THROW_ON_ERROR);
+            $this->assertSame($deleted, $report['audit_events_deleted']);
+            $this->assertSame($remaining, $report['retained_events']);
+            $this->assertSame(3, $report['configured_limit']);
+            $this->assertSame($remaining, WorkflowScheduleHistoryEvent::query()
+                ->where('workflow_schedule_id', $schedule->id)
+                ->count());
+        }
+        $this->assertSame([1, 205, 206], WorkflowScheduleHistoryEvent::query()
+            ->where('workflow_schedule_id', $schedule->id)
+            ->orderBy('sequence')
+            ->pluck('sequence')
+            ->all());
+    }
+
+    public function testCanonicalEnvironmentLimitIsParsedAndOverflowRefuses(): void
+    {
+        config()->set('workflows.v2.schedules.audit_history_limit', '3');
+        $this->assertSame(3, ScheduleAuditRetention::limit());
+        config()
+            ->set('workflows.v2.schedules.audit_history_limit', (string) PHP_INT_MAX);
+        $this->assertSame(PHP_INT_MAX, ScheduleAuditRetention::limit());
+        config()
+            ->set('workflows.v2.schedules.audit_history_limit', PHP_INT_MAX . '0');
+        $this->expectException(InvalidArgumentException::class);
+        ScheduleAuditRetention::limit();
+    }
+
+    public function testCompactingAManualTriggerDoesNotInventAnOccurrenceReceipt(): void
+    {
+        config()->set('workflows.v2.schedules.audit_history_limit', null);
+        $schedule = $this->schedule('manual');
+        $manual = WorkflowScheduleHistoryEvent::record($schedule, HistoryEventType::ScheduleTriggered);
+        WorkflowScheduleHistoryEvent::record($schedule, HistoryEventType::ScheduleUpdated);
+        config()
+            ->set('workflows.v2.schedules.audit_history_limit', 2);
+        $this->assertSame(1, ScheduleAuditRetention::prune($schedule));
+        $this->assertModelMissing($manual);
+        $this->assertSame(0, WorkflowScheduleOccurrence::query()->count());
+        $this->assertSame([1, 3], $schedule->historyEvents()->orderBy('sequence')->pluck('sequence')->all());
+    }
+
+    public static function unusableLegacyClocks(): array
+    {
+        return [
+            'non-string clock' => [123],
+            'unsupported clock syntax' => ['not-a-clock'],
+            'invalid calendar day' => ['2026-02-30T01:02:03Z'],
+        ];
+    }
+
+    #[DataProvider('unusableLegacyClocks')]
+    public function testUnusableLegacyPayloadRefusesBothCompactionAndTheCommand(mixed $clock): void
+    {
+        config()->set('workflows.v2.schedules.audit_history_limit', null);
+        $schedule = $this->schedule('legacy-refusal', 'orders');
+        $legacy = WorkflowScheduleHistoryEvent::record($schedule, HistoryEventType::ScheduleTriggered, [
+            'occurrence_time' => '2026-11-13T01:02:03Z',
+        ]);
+        $payload = $legacy->payload;
+        $payload['occurrence_time'] = $clock;
+        $legacy->forceFill([
+            'occurrence_at_utc' => null,
+            'payload' => $payload,
+        ])->save();
+        WorkflowScheduleOccurrence::query()->delete();
+        WorkflowScheduleHistoryEvent::record($schedule, HistoryEventType::ScheduleUpdated);
+        $before = $schedule->historyEvents()
+            ->orderBy('sequence')
+            ->get()
+            ->toArray();
+        config()
+            ->set('workflows.v2.schedules.audit_history_limit', 2);
+        try {
+            ScheduleAuditRetention::prune($schedule);
+            $this->fail('Unusable legacy payload was discarded.');
+        } catch (LogicException $error) {
+            $this->assertStringContainsString('legacy schedule occurrence cannot be established', $error->getMessage());
+        }
+        $this->assertSame($before, $schedule->historyEvents()->orderBy('sequence')->get()->toArray());
+        $this->assertSame(1, Artisan::call('workflow:v2:schedule-audit-prune', [
+            '--namespace' => 'orders',
+            '--schedule' => 'legacy-refusal',
+        ]));
+        $this->assertStringContainsString('audit row was retained', Artisan::output());
+        $this->assertSame($before, $schedule->historyEvents()->orderBy('sequence')->get()->toArray());
+        $this->assertSame(0, WorkflowScheduleOccurrence::query()->count());
+    }
+
+    public function testAConflictingReceiptNamespaceRollsBackTheAuditAppend(): void
+    {
+        $schedule = $this->schedule('conflicting-receipt', 'orders');
+        $receipt = WorkflowScheduleOccurrence::query()->create([
+            'workflow_schedule_id' => $schedule->id,
+            'namespace' => 'other',
+            'occurrence_at_utc' => '2026-11-13 01:02:03.000000',
+        ]);
+        $before = $schedule->historyEvents()
+            ->get()
+            ->toArray();
+        try {
+            WorkflowScheduleHistoryEvent::record($schedule, HistoryEventType::ScheduleTriggered, [
+                'occurrence_time' => '2026-11-13T01:02:03Z',
+            ]);
+            $this->fail('An occurrence receipt with another namespace was reused.');
+        } catch (LogicException $error) {
+            $this->assertStringContainsString('conflicting namespace ownership', $error->getMessage());
+        }
+        $this->assertSame($before, $schedule->historyEvents()->get()->toArray());
+        $this->assertSame('other', $receipt->fresh()->namespace);
+        $this->assertSame(1, WorkflowScheduleOccurrence::query()->count());
+    }
+
+    public function testTheCleanupApiRejectsOutOfRangeLimitsWithoutDeletingOrphans(): void
+    {
+        $schedule = $this->schedule('invalid-api-limit', 'orders');
+        $schedule->delete();
+        foreach ([0, 1001] as $limit) {
+            try {
+                ScheduleAuditRetention::cleanup('orders', $limit);
+                $this->fail('An invalid cleanup limit was accepted.');
+            } catch (InvalidArgumentException $error) {
+                $this->assertStringContainsString('between 1 and 1000', $error->getMessage());
+            }
+            $this->assertSame(1, $schedule->historyEvents()->count());
+        }
+    }
+
+    public static function invalidCommandNamespaces(): array
+    {
+        return [
+            'empty' => [''],
+            'too long' => [str_repeat('x', 256)],
+            'invalid UTF-8' => ["\xff"],
+        ];
+    }
+
+    #[DataProvider('invalidCommandNamespaces')]
+    public function testAnInvalidCommandNamespaceCannotDeleteRows(string $namespace): void
+    {
+        $schedule = $this->schedule('invalid-namespace', 'orders');
+        $schedule->delete();
+        $this->assertSame(1, Artisan::call('workflow:v2:schedule-audit-prune', [
+            '--namespace' => $namespace,
+        ]));
+        $this->assertStringContainsString('namespace must be a nonempty UTF-8 string', Artisan::output());
+        $this->assertSame(1, $schedule->historyEvents()->count());
+    }
+
+    public function testARequestedScheduleMustExistWithItsExactIdInTheSelectedNamespace(): void
+    {
+        $schedule = $this->schedule('exact-schedule', 'orders');
+        $before = $schedule->historyEvents()
+            ->get()
+            ->toArray();
+        foreach ([['other', 'exact-schedule'],
+            ['orders', 'missing'],
+            ['orders', 'EXACT-SCHEDULE'],
+        ] as [$namespace, $id]) {
+            $this->assertSame(1, Artisan::call('workflow:v2:schedule-audit-prune', [
+                '--namespace' => $namespace,
+                '--schedule' => $id,
+            ]));
+            $this->assertStringContainsString('exact schedule ID does not exist', Artisan::output());
+            $this->assertSame($before, $schedule->historyEvents()->get()->toArray());
+        }
+    }
+
+    public function testTheOccurrenceMigrationIsIdempotentWithoutDiscardingReceipts(): void
+    {
+        $schedule = $this->schedule('migration-repeat');
+        WorkflowScheduleHistoryEvent::record($schedule, HistoryEventType::ScheduleTriggered, [
+            'occurrence_time' => '2026-11-13T01:02:03Z',
+        ]);
+        $before = WorkflowScheduleOccurrence::query()->get()->toArray();
+        $migration = require __DIR__ . '/../../../src/migrations/2026_11_14_000200_create_workflow_schedule_occurrences.php';
+        $migration->up();
+        $this->assertSame($before, WorkflowScheduleOccurrence::query()->get()->toArray());
+    }
+
+    public function testAnEmptyOccurrenceMigrationCanRollBackAndRepeatWithoutAnExistingTable(): void
+    {
+        $migration = require __DIR__ . '/../../../src/migrations/2026_11_14_000200_create_workflow_schedule_occurrences.php';
+        $this->assertSame(0, WorkflowScheduleOccurrence::query()->count());
+        try {
+            $migration->down();
+            $this->assertFalse(Schema::hasTable('workflow_schedule_occurrences'));
+            $migration->down();
+            $this->assertFalse(Schema::hasTable('workflow_schedule_occurrences'));
+        } finally {
+            $migration->up();
+        }
+        $this->assertTrue(Schema::hasTable('workflow_schedule_occurrences'));
+        $this->assertSame(0, WorkflowScheduleOccurrence::query()->count());
+    }
+
+    public function testMigrationCannotDiscardPopulatedOccurrenceReceipts(): void
+    {
+        $schedule = $this->schedule('rollback');
+        WorkflowScheduleHistoryEvent::record($schedule, HistoryEventType::ScheduleTriggered, [
+            'occurrence_time' => '2026-11-13T01:02:03Z',
+        ]);
+        $migration = require __DIR__ . '/../../../src/migrations/2026_11_14_000200_create_workflow_schedule_occurrences.php';
+        try {
+            $migration->down();
+            $this->fail('The deduplication receipt table was dropped.');
+        } catch (LogicException $error) {
+            $this->assertStringContainsString('receipts must be preserved', $error->getMessage());
+        }
+        $this->assertTrue(Schema::hasTable('workflow_schedule_occurrences'));
+        $this->assertSame(1, WorkflowScheduleOccurrence::query()->count());
+    }
+
+    private function schedule(string $id, ?string $namespace = null): WorkflowSchedule
+    {
+        return ScheduleManager::create(
+            scheduleId: $id,
+            workflowClass: TestScheduledWorkflow::class,
+            cronExpression: '* * * * *',
+            overlapPolicy: ScheduleOverlapPolicy::AllowAll,
+            namespace: $namespace,
+        );
+    }
+}
+
+final class HiddenScheduleForAuditRetention extends WorkflowSchedule
+{
+    protected static function booted(): void
+    {
+        static::addGlobalScope('application-filter', static function (Builder $query): void {
+            $query->whereRaw('1=0');
+        });
+    }
+}
+
+final class HiddenOccurrenceForAuditRetention extends WorkflowScheduleOccurrence
+{
+    protected static function booted(): void
+    {
+        static::addGlobalScope('application-filter', static function (Builder $query): void {
+            $query->whereRaw('1=0');
+        });
+    }
+}
+
+final class HiddenScheduleHistoryForAuditRetention extends WorkflowScheduleHistoryEvent
+{
+    protected static function booted(): void
+    {
+        static::addGlobalScope('application-filter', static function (Builder $query): void {
+            $query->whereRaw('1=0');
+        });
+    }
+}
diff --git a/tests/Feature/V2/V2ScheduleTest.php b/tests/Feature/V2/V2ScheduleTest.php
index d571acbae..0a39e879f 100644
--- a/tests/Feature/V2/V2ScheduleTest.php
+++ b/tests/Feature/V2/V2ScheduleTest.php
@@ -1126,10 +1126,8 @@ public function testScheduleHistorySequenceRecoversFromConcurrentCollision(): vo
             cronExpression: '* * * * *',
         );
 
-        // ScheduleCreated was written with sequence = 1. Simulate another
-        // concurrent writer that has already claimed sequence = 2 before we
-        // attempt to record the next lifecycle event — exactly the collision
-        // the unique(workflow_schedule_id, sequence) constraint would raise.
+        // An already committed sibling advances the initial maximum read.
+        // The independent-writer test below exercises the actual collision.
         WorkflowScheduleHistoryEvent::query()->create([
             'workflow_schedule_id' => $schedule->id,
             'schedule_id' => $schedule->schedule_id,
@@ -1149,8 +1147,7 @@ public function testScheduleHistorySequenceRecoversFromConcurrentCollision(): vo
             ->orderBy('sequence')
             ->get();
 
-        // record() should have retried past the collision and landed on
-        // sequence = 3, preserving ordering rather than blowing up.
+        // The new append follows the committed sibling without reusing its slot.
         $this->assertSame([1, 2, 3], $events->pluck('sequence')->all());
         $this->assertSame(
             [
@@ -1173,9 +1170,7 @@ public function testConcurrentScheduleLifecycleWritesAllLandWithUniqueSequences(
             cronExpression: '* * * * *',
         );
 
-        // Pre-claim sequences 2..6 to force record() to retry five times
-        // before landing on the next free slot. This exercises the retry
-        // loop under a larger pileup than a single collision.
+        // A committed pileup advances the first allocation to the next free slot.
         for ($claimed = 2; $claimed <= 6; $claimed++) {
             WorkflowScheduleHistoryEvent::query()->create([
                 'workflow_schedule_id' => $schedule->id,
@@ -1208,79 +1203,157 @@ public function testConcurrentScheduleLifecycleWritesAllLandWithUniqueSequences(
         $this->assertSame('post-pileup', $tail->payload['reason']);
     }
 
-    public function testRecordRetriesWhenStaleMaxSequenceCollidesWithUniqueIndex(): void
+    public function testScheduleAuditHandlesIndependentSequenceContention(): void
     {
-        // Prior regression coverage (the two tests above) exercised the happy
-        // path where record() reads a fresh max(sequence) and lands on the
-        // next free slot. They do NOT hit the UniqueConstraintViolationException
-        // branch of record(), because by the time record() runs the
-        // pre-inserted rows have already pushed max() up.
-        //
-        // This test forces the collision path: we hook the Eloquent `creating`
-        // event so the *first* create() call inside record() runs into a
-        // sibling row at the same sequence, racing the unique index. The hook
-        // fires exactly once — record()'s retry loop must then re-read max()
-        // and insert at the next free slot.
-
         $schedule = ScheduleManager::create(
             scheduleId: 'stale-max-race-test',
             workflowClass: TestScheduledWorkflow::class,
             cronExpression: '* * * * *',
         );
-
-        $raced = false;
-
-        WorkflowScheduleHistoryEvent::creating(static function (WorkflowScheduleHistoryEvent $event) use (
-            &$raced,
-            $schedule
-        ): void {
-            // Only race the very first attempt inside record(). The retry
-            // re-reads max(sequence) and should land on the next slot, which
-            // this hook no longer interferes with.
-            if ($raced || (int) $event->sequence !== 2) {
-                return;
-            }
-
-            $raced = true;
-
-            // Commit a sibling row at the exact sequence record() is about to
-            // try. On MySQL/Postgres this produces UniqueConstraintViolationException
-            // when record()'s create() runs; on SQLite the same unique-index
-            // violation is raised by Laravel as UniqueConstraintViolationException.
-            WorkflowScheduleHistoryEvent::query()->create([
-                'workflow_schedule_id' => $schedule->id,
-                'schedule_id' => $schedule->schedule_id,
-                'namespace' => $schedule->namespace,
-                'sequence' => 2,
-                'event_type' => HistoryEventType::SchedulePaused->value,
-                'payload' => [
-                    'reason' => 'raced the pending writer',
-                ],
-                'recorded_at' => now(),
-            ]);
-        });
-
+        $original = $schedule->getConnection();
+        $driver = $original->getDriverName();
+        $this->assertSame(0, $original->transactionLevel());
+        $databaseFile = null;
+        $primaryName = 'schedule_audit_primary';
+        $competingName = 'schedule_audit_competing';
+        $storageConnection = config('workflows.storage.connection');
+        $modelConnection = $schedule->getConnectionName();
+        $dispatcher = WorkflowScheduleHistoryEvent::getEventDispatcher();
+        $this->assertNotNull($dispatcher);
         try {
+            $configuration = $original->getConfig();
+            unset($configuration['name'], $configuration['read'], $configuration['write'], $configuration['url']);
+            $configuration['options'][\PDO::ATTR_PERSISTENT] = false;
+            if ($driver === 'sqlite') {
+                // An in-memory fixture cannot provide a genuine second writer.
+                // Copy its logical state into an owned file without modifying it.
+                $createdFile = tempnam(sys_get_temp_dir(), 'schedule-audit-race-');
+                $this->assertNotFalse($createdFile);
+                $databaseFile = $createdFile;
+                $original->statement('VACUUM INTO ?', [$databaseFile]);
+                $configuration['database'] = $databaseFile;
+            }
+            config()
+                ->set('database.connections.' . $primaryName, $configuration);
+            config()
+                ->set('database.connections.' . $competingName, $configuration);
+            $primary = DB::connection($primaryName);
+            $competing = DB::connection($competingName);
+            $this->assertNotSame($primary->getPdo(), $competing->getPdo());
+            if (in_array($driver, ['mysql', 'mariadb'], true)) {
+                $primary->statement('SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ');
+                $competing->statement('SET SESSION innodb_lock_wait_timeout = 1');
+                $this->assertNotSame(
+                    $primary->scalar('SELECT CONNECTION_ID()'),
+                    $competing->scalar('SELECT CONNECTION_ID()')
+                );
+            } elseif ($driver === 'pgsql') {
+                $primary->statement('SET SESSION CHARACTERISTICS AS TRANSACTION ISOLATION LEVEL READ COMMITTED');
+                $competing->statement("SET lock_timeout = '1s'");
+                $this->assertNotSame(
+                    $primary->scalar('SELECT pg_backend_pid()'),
+                    $competing->scalar('SELECT pg_backend_pid()')
+                );
+            } elseif ($driver === 'sqlite') {
+                $competing->statement('PRAGMA busy_timeout = 0');
+            }
+            config()
+                ->set('workflows.storage.connection', $primaryName);
+            $schedule->setConnection($primaryName);
+            WorkflowScheduleHistoryEvent::setEventDispatcher(clone $dispatcher);
+            $raced = false;
+            $committed = false;
+            $blocked = false;
+            $attempts = 0;
+            WorkflowScheduleHistoryEvent::creating(function (WorkflowScheduleHistoryEvent $event) use (
+                &$raced,
+                &$committed,
+                &$blocked,
+                &$attempts,
+                $schedule,
+                $competing,
+                $driver
+            ): void {
+                if ($event->workflow_schedule_id !== $schedule->id) {
+                    return;
+                }
+                ++$attempts;
+                if ($raced || (int) $event->sequence !== 2) {
+                    return;
+                }
+                $raced = true;
+                $this->assertGreaterThan(0, $event->getConnection()->transactionLevel());
+                $this->assertSame(0, $competing->transactionLevel());
+                try {
+                    $competing->transaction(static function () use ($competing, $event): void {
+                        $competing->table($event->getTable())
+                            ->insert([
+                                'id' => (string) Str::ulid(),
+                                'workflow_schedule_id' => $event->workflow_schedule_id,
+                                'schedule_id' => $event->schedule_id,
+                                'namespace' => $event->namespace,
+                                'sequence' => 2,
+                                'event_type' => HistoryEventType::SchedulePaused->value,
+                                'payload' => json_encode([
+                                    'reason' => 'independently committed sibling',
+                                ], JSON_THROW_ON_ERROR),
+                                'recorded_at' => now()
+                                    ->format('Y-m-d H:i:s.u'),
+                                'created_at' => now()
+                                    ->format('Y-m-d H:i:s.u'),
+                                'updated_at' => now()
+                                    ->format('Y-m-d H:i:s.u'),
+                            ]);
+                    });
+                    $committed = true;
+                } catch (\Illuminate\Database\QueryException $error) {
+                    if ($driver !== 'sqlite') {
+                        throw $error;
+                    }
+                    $this->assertContains(((int) ($error->errorInfo[1] ?? 0)) & 255, [5, 6]);
+                    $blocked = true;
+                }
+                $this->assertSame(0, $competing->transactionLevel());
+            });
             ScheduleManager::pause($schedule, 'after retry');
+            $events = WorkflowScheduleHistoryEvent::query()
+                ->where('workflow_schedule_id', $schedule->id)
+                ->orderBy('sequence')
+                ->get();
+            $this->assertTrue($raced, 'The independent writer was not exercised.');
+            if ($driver === 'sqlite') {
+                // Writer admission prevents an independent SQLite collision.
+                $this->assertTrue($blocked);
+                $this->assertFalse($committed);
+                $this->assertSame(1, $attempts);
+                $this->assertSame([1, 2], $events->pluck('sequence')->all());
+                $this->assertSame('after retry', $events[1]->payload['reason']);
+            } else {
+                // The sibling commit survives our failed insert/savepoint.
+                $this->assertTrue($committed);
+                $this->assertFalse($blocked);
+                $this->assertSame(2, $attempts);
+                $this->assertSame([1, 2, 3], $events->pluck('sequence')->all());
+                $this->assertSame('independently committed sibling', $events[1]->payload['reason']);
+                $this->assertSame('after retry', $events[2]->payload['reason']);
+            }
+            $this->assertSame(HistoryEventType::ScheduleCreated, $events[0]->event_type);
+            $this->assertSame(HistoryEventType::SchedulePaused, $events->last()->event_type);
         } finally {
-            WorkflowScheduleHistoryEvent::flushEventListeners();
+            WorkflowScheduleHistoryEvent::setEventDispatcher($dispatcher);
+            config()
+                ->set('workflows.storage.connection', $storageConnection);
+            $schedule->setConnection($modelConnection);
+            DB::purge($competingName);
+            DB::purge($primaryName);
+            if ($databaseFile !== null) {
+                foreach (['', '-journal', '-wal', '-shm'] as $suffix) {
+                    if (is_file($databaseFile . $suffix)) {
+                        unlink($databaseFile . $suffix);
+                    }
+                }
+            }
         }
-
-        $this->assertTrue($raced, 'The creating-hook race was not exercised; test fixture is broken.');
-
-        $events = WorkflowScheduleHistoryEvent::query()
-            ->where('workflow_schedule_id', $schedule->id)
-            ->orderBy('sequence')
-            ->get();
-
-        // record() must land the SchedulePaused event at sequence 3 after the
-        // retry, preserving monotonic ordering without a duplicate-key fatal.
-        $this->assertSame([1, 2, 3], $events->pluck('sequence')->all());
-
-        $this->assertSame(HistoryEventType::ScheduleCreated->value, $events[0]->event_type->value);
-        $this->assertSame('raced the pending writer', $events[1]->payload['reason']);
-        $this->assertSame(HistoryEventType::SchedulePaused->value, $events[2]->event_type->value);
-        $this->assertSame('after retry', $events[2]->payload['reason']);
     }
 
     public function testTriggerDeletedScheduleTracksSkipReason(): void