Keep v1 projections read-only and bound import history

app/Http/Controllers/Api/MessageStreamController.php

@@ -4,6 +4,7 @@
 
 use App\Support\AvroPayloadEnvelopeResolver;
 use App\Support\ControlPlaneProtocol;
+use App\Support\LegacyV1Projection;
 use App\Support\MessageStreamService;
 use App\Support\NamespaceExternalPayloadStorage;
 use App\Support\NamespaceWorkflowScope;
@@ -25,7 +26,7 @@
         private readonly WorkflowCommandContextFactory $commandContexts,
     ) {}
 
     public function append(Request $request, string $workflowId, string $streamName): JsonResponse
     {
         if ($response = ControlPlaneProtocol::rejectUnsupported($request)) {
             return $response;
@@ -44,6 +45,15 @@
             ], 404);
         }
 
+        $run = NamespaceWorkflowScope::currentRun($namespace, $workflowId);
+        if (LegacyV1Projection::isProjectedRun($run)) {
+            return ControlPlaneProtocol::jsonForRequest(
+                $request,
+                LegacyV1Projection::readOnlyMessageAppend($run, $streamName),
+                409,
+            );
+        }
+
         $validator = Validator::make($request->all(), [
             'message_id' => ['required', 'string', 'max:191', 'regex:/^[A-Za-z0-9._:-]+$/'],
             'input' => ['nullable', 'array'],

app/Http/Controllers/Api/WorkflowController.php

@@ -1319,18 +1319,7 @@ private function rejectProjectedV1Operation(
 
     private function projectedV1OperationResponse(Request $request, WorkflowRun $run, string $operation): JsonResponse
     {
-        return ControlPlaneProtocol::jsonForRequest($request, [
-            'message' => sprintf(
-                'Operation [%s] is unavailable because this execution is a read-only projection owned by the source v1 runtime.',
-                $operation,
-            ),
-            'reason' => 'v1_projection_read_only',
-            'remediation' => 'Run commands against the source v1 application until it completes; repeat projection never dispatches or advances standalone execution.',
-            'workflow_id' => $run->workflow_instance_id,
-            'run_id' => $run->id,
-            'operation' => $operation,
-            'execution_owner' => 'v1',
-        ], 409);
+        return ControlPlaneProtocol::jsonForRequest($request, LegacyV1Projection::readOnlyOperation($run, $operation), 409);
     }
 
     /**

app/Support/LegacyV1Projection.php

@@ -10,15 +10,17 @@
 use Workflow\V2\Models\WorkflowInstance;
 use Workflow\V2\Models\WorkflowRun;
 use Workflow\V2\Models\WorkflowRunSummary;
+use Workflow\V2\Support\HistoryBudget;
+use Workflow\V2\Support\StructuralLimits;
 
 final class LegacyV1Projection
 {
     private const OPAQUE_PAYLOAD_REASON = 'legacy_php_serialization_not_portable';
 
     private const OPAQUE_PAYLOAD_REMEDIATION = 'Treat the preserved value as opaque; export decoded JSON from the source application if portable payload access is required.';
 
     /**
-     * @param array<string, mixed> $source
+     * @param  array<string, mixed>  $source
      * @return array<string, mixed>
      */
     public static function import(string $sourceId, array $source, ?string $namespace, bool $dryRun = false): array
@@ -112,10 +114,7 @@
                 $events = $normalized['events'];
                 $status = $normalized['status'];
                 $closedAt = $normalized['closed_at'];
-                $historySize = array_sum(array_map(
-                    static fn (array $event): int => strlen((string) json_encode($event, JSON_UNESCAPED_SLASHES)),
-                    $events,
-                ));
+                $historySize = $normalized['history_size_bytes'];
 
                 WorkflowInstance::query()->create([
                     'id' => $workflowId,
@@ -233,6 +232,35 @@
             && $run->import_source === LegacyV1ProjectionContract::IMPORT_SOURCE;
     }
 
+    /** @return array<string, mixed> */
+    public static function readOnlyOperation(WorkflowRun $run, string $operation): array
+    {
+        return [
+            'message' => sprintf(
+                'Operation [%s] is unavailable because this execution is a read-only projection owned by the source v1 runtime.',
+                $operation,
+            ),
+            'reason' => 'v1_projection_read_only',
+            'remediation' => 'Run commands against the source v1 application until it completes; repeat projection never dispatches or advances standalone execution.',
+            'workflow_id' => $run->workflow_instance_id,
+            'run_id' => $run->id,
+            'operation' => $operation,
+            'execution_owner' => 'v1',
+        ];
+    }
+
+    /** @return array<string, mixed> */
+    public static function readOnlyMessageAppend(WorkflowRun $run, string $streamName): array
+    {
+        return [
+            ...self::readOnlyOperation($run, 'message_stream_append'),
+            'accepted' => false,
+            'duplicate' => false,
+            'stream_name' => $streamName,
+            'outcome' => 'rejected',
+        ];
+    }
+
     /**
      * @return array<string, mixed>|null
      */
@@ -274,7 +302,7 @@
      * Add provenance to a history bundle and refresh its checksum so the
      * exported document remains self-consistent.
      *
-     * @param array<string, mixed> $bundle
+     * @param  array<string, mixed>  $bundle
      * @return array<string, mixed>
      */
     public static function decorateHistoryExport(array $bundle, WorkflowRun $run): array
@@ -333,10 +361,10 @@
     }
 
     /**
-     * @param array<string, mixed> $source
+     * @param  array<string, mixed>  $source
      * @return array<string, mixed>
      */
     private static function normalize(string $sourceId, array $source, ?string $namespace): array
     {
         $sourceId = trim($sourceId);
         $namespace = is_string($namespace) && trim($namespace) !== '' ? trim($namespace) : 'default';
@@ -375,6 +403,26 @@
             ];
         }
 
+        $terminal = in_array($status, ['completed', 'failed', 'cancelled', 'terminated'], true);
+        $eventCount = 1 + (int) $terminal;
+        foreach (['logs', 'signals', 'exceptions'] as $collection) {
+            if (! is_array($source[$collection] ?? null)) {
+                continue;
+            }
+            foreach ($source[$collection] as $row) {
+                $eventCount += (int) is_array($row);
+            }
+        }
+        $eventLimit = StructuralLimits::historyTransactionSizeLimit();
+        if ($eventLimit > 0 && $eventCount > $eventLimit) {
+            return [
+                'valid' => false,
+                'namespace' => $namespace,
+                'errors' => [self::limitError('history_transaction_size', $eventCount, $eventLimit)],
+                'unsupported_fields' => [],
+            ];
+        }
+
         $mapped = self::mappedIdentity($sourceId, $qualifiedId);
         $source['class'] = $class;
         $source['connection'] = self::string($source['connection'] ?? null);
@@ -410,7 +458,6 @@
         ];
         $startedAt = self::timestamp($source['created_at'] ?? null);
         $lastProgressAt = self::timestamp($source['updated_at'] ?? null) ?? $startedAt;
-        $terminal = in_array($status, ['completed', 'failed', 'cancelled', 'terminated'], true);
         $closedAt = $terminal ? $lastProgressAt : null;
         $normalizedForHash = [
             'origin' => $origin,
@@ -427,13 +474,22 @@
             'created_at' => self::timestampString($startedAt),
             'updated_at' => self::timestampString($lastProgressAt),
         ];
-        $dedupeKey = hash('sha256', json_encode(self::canonicalize($normalizedForHash), JSON_UNESCAPED_SLASHES | JSON_THROW_ON_ERROR));
         $migration = [
             'origin' => $origin,
             'identity' => $identity,
             'task_queue_context' => $taskQueue,
             'unsupported_fields' => $unsupported,
         ];
+        $built = self::events($source, $migration, $status, $startedAt, $lastProgressAt, $mapped['run_id']);
+        if (isset($built['limit_error'])) {
+            return [
+                'valid' => false,
+                'namespace' => $namespace,
+                'errors' => [$built['limit_error']],
+                'unsupported_fields' => [],
+            ];
+        }
+        $dedupeKey = hash('sha256', json_encode(self::canonicalize($normalizedForHash), JSON_UNESCAPED_SLASHES | JSON_THROW_ON_ERROR));
 
         return [
             'valid' => true,
@@ -451,14 +507,15 @@
             'import_id' => $mapped['import_id'],
             'dedupe_key' => $dedupeKey,
             'unsupported_fields' => $unsupported,
-            'events' => self::events($source, $migration, $status, $startedAt, $lastProgressAt, $mapped['run_id']),
+            'events' => $built['events'],
+            'history_size_bytes' => $built['history_size_bytes'],
         ];
     }
 
     /**
-     * @param array<string, mixed> $source
-     * @param array<string, mixed> $migration
-     * @return list<array<string, mixed>>
+     * @param  array<string, mixed>  $source
+     * @param  array<string, mixed>  $migration
+     * @return array{events: list<array<string, mixed>>, history_size_bytes: int}|array{limit_error: array<string, mixed>}
      */
     private static function events(array $source, array $migration, string $status, ?Carbon $startedAt, ?Carbon $lastProgressAt, string $runId): array
     {
@@ -467,7 +524,7 @@
             'recorded_at' => $startedAt,
             'boundary' => 0,
             'order' => 0,
-            'payload' => [
+            'build_payload' => static fn (): array => [
                 'migration_projection' => $migration,
                 'legacy_payloads' => [
                     'arguments' => self::opaquePayload($source['arguments'] ?? null),
@@ -481,7 +538,7 @@
                 'recorded_at' => self::timestamp($log['created_at'] ?? $log['now'] ?? null),
                 'boundary' => 1,
                 'order' => 100 + $index,
-                'payload' => [
+                'build_payload' => static fn (): array => [
                     'migration_projection' => [
                         'source_record' => 'workflow_log',
                         'legacy_id' => $log['id'] ?? null,
@@ -502,7 +559,7 @@
                 'recorded_at' => self::timestamp($signal['created_at'] ?? $signal['received_at'] ?? null),
                 'boundary' => 1,
                 'order' => 200 + $index,
-                'payload' => [
+                'build_payload' => static fn (): array => [
                     'migration_projection' => [
                         'source_record' => 'workflow_signal',
                         'legacy_id' => $signal['id'] ?? null,
@@ -519,7 +576,7 @@
                 'recorded_at' => self::timestamp($exception['created_at'] ?? null),
                 'boundary' => 1,
                 'order' => 300 + $index,
-                'payload' => [
+                'build_payload' => static fn (): array => [
                     'migration_projection' => [
                         'source_record' => 'workflow_exception',
                         'legacy_id' => $exception['id'] ?? null,
@@ -545,7 +602,7 @@
                 'recorded_at' => $lastProgressAt,
                 'boundary' => 2,
                 'order' => PHP_INT_MAX,
-                'payload' => [
+                'build_payload' => static fn (): array => [
                     'migration_projection' => [
                         'source_status' => $source['status'] ?? null,
                         'execution_owner' => 'v1',
@@ -564,19 +621,42 @@
             return [$left['boundary'], $leftTime, $left['order']] <=> [$right['boundary'], $rightTime, $right['order']];
         });
 
+        $historySize = 0;
+        $byteLimit = StructuralLimits::historyTransactionBytesLimit();
         foreach ($events as $index => &$event) {
+            $buildPayload = $event['build_payload'];
+            unset($event['boundary'], $event['order'], $event['build_payload']);
+            $event['payload'] = $buildPayload();
             $sequence = $index + 1;
             $event['sequence'] = $sequence;
             $event['id'] = substr(hash('sha256', $runId."\0".$sequence."\0".$event['event_type']), 0, 26);
-            unset($event['boundary'], $event['order']);
+            $historySize += HistoryBudget::sizeOfEvent(new WorkflowHistoryEvent($event));
+            if ($byteLimit > 0 && $historySize > $byteLimit) {
+                return ['limit_error' => self::limitError('history_transaction_bytes', $historySize, $byteLimit)];
+            }
         }
         unset($event);
 
-        return $events;
+        return ['events' => $events, 'history_size_bytes' => $historySize];
+    }
+
+    /** @return array<string, mixed> */
+    private static function limitError(string $dimension, int $used, int $limit): array
+    {
+        [$measure, $setting] = $dimension === 'history_transaction_size'
+            ? ['history events', 'DW_V2_LIMIT_HISTORY_TRANSACTION_SIZE']
+            : ['history bytes', 'DW_V2_LIMIT_HISTORY_TRANSACTION_BYTES'];
+
+        return [
+            'reason' => 'v1_projection_too_large',
+            'message' => "The projection would write {$used} {$measure} in one transaction; at most {$limit} are allowed ({$dimension}).",
+            'remediation' => "Keep the workflow on its v1 runtime, or raise {$setting} on this server before importing it.",
+            'limit' => compact('dimension', 'used', 'limit'),
+        ];
     }
 
     /**
-     * @param array<string, mixed> $source
+     * @param  array<string, mixed>  $source
      * @return list<array{field: string, reason: string, remediation: string}>
      */
     private static function unsupportedFields(array $source, string $status): array
@@ -652,16 +732,16 @@
     }
 
     /**
-     * @param array<string, mixed> $normalized
-     * @param array<string, mixed> $extra
+     * @param  array<string, mixed>  $normalized
+     * @param  array<string, mixed>  $extra
      * @return array<string, mixed>
      */
     private static function report(array $normalized, string $status, array $extra = []): array
     {
         $errors = is_array($normalized['errors'] ?? null) ? array_values($normalized['errors']) : [];
         $rejection = [];
         if ($status === 'rejected' && is_array($errors[0] ?? null)) {
-            $rejection = array_intersect_key($errors[0], array_flip(['reason', 'message', 'remediation']));
+            $rejection = array_intersect_key($errors[0], array_flip(['reason', 'message', 'remediation', 'limit']));
         }
 
         return [

app/Support/MessageStreamService.php

@@ -56,6 +56,9 @@ public function append(
             }
 
             $run = NamespaceWorkflowScope::currentRun($namespace, $workflowId);
+            if (LegacyV1Projection::isProjectedRun($run)) {
+                return ['status' => 409, ...LegacyV1Projection::readOnlyMessageAppend($run, $streamName)];
+            }
             if (! $run instanceof WorkflowRun || $run->status->isTerminal()) {
                 return $this->rejection('run_not_active', 409, $workflowId, $streamName, $messageId);
             }
@@ -168,7 +171,7 @@ public function recordMalformed(
             }
 
             $run = NamespaceWorkflowScope::currentRun($namespace, $workflowId);
-            if (! $run instanceof WorkflowRun || $run->status->isTerminal()) {
+            if (! $run instanceof WorkflowRun || LegacyV1Projection::isProjectedRun($run) || $run->status->isTerminal()) {
                 return;
             }
 

config/dw-contract.php

@@ -1013,6 +1013,12 @@
             'since' => '2.0.0',
             'legacy' => 'WORKFLOW_V2_LIMIT_HISTORY_TRANSACTION_SIZE',
         ],
+        'DW_V2_LIMIT_HISTORY_TRANSACTION_BYTES' => [
+            'description' => 'Package-level history-transaction byte ceiling, also applied to Waterline v1 projection imports. Zero disables this dimension.',
+            'default' => '16777216',
+            'since' => '2.11.5',
+            'legacy' => 'WORKFLOW_V2_LIMIT_HISTORY_TRANSACTION_BYTES',
+        ],
         'DW_V2_LIMIT_WARNING_THRESHOLD_PERCENT' => [
             'description' => 'Percent of a structural limit at which the package emits an approaching-limit warning.',
             'default' => '80',

docs/bounded-growth.md

@@ -58,6 +58,28 @@ retention-deleted history. Retention and rebuild maintenance should leave a
 marked run without recreating its deleted summary. Check `/api/system/health`,
 `/api/system/metrics` and `/api/system/operator-metrics` after an upgrade.
 
+## Waterline v1 projection imports
+
+A projection import writes its history in one transaction. It uses
+`DW_V2_LIMIT_HISTORY_TRANSACTION_SIZE` (5,000 events by default) and
+`DW_V2_LIMIT_HISTORY_TRANSACTION_BYTES` (16 MiB by default). Setting either
+dimension to `0` disables that dimension. These limits supplement the HTTP
+request payload limit.
+
+The event count includes the start event, each log, signal and exception row,
+and a terminal event when applicable. The byte budget and stored summary use
+Core's canonical history metric: the event type's byte length plus its payload's
+JSON byte length, with unescaped slashes and Unicode. The server checks the count
+before copying or hashing source rows, then builds payloads in history order and stops at the
+first event exceeding the byte budget.
+
+An oversized import returns HTTP 409 with `reason: v1_projection_too_large`
+and `limit: {dimension, used, limit}` before reading existing workflow rows or
+starting the write transaction. Dry runs enforce the same limits. Keep the
+workflow on its v1 runtime or deliberately raise the reported limit before
+importing it. Imported projections remain read-only, including message-stream
+appends and malformed-input accounting.
+
 ## Cache Inventory
 
 | Policy ID | Prefix | Owner | Growth Bound |

docs/server-reference.md

@@ -1930,6 +1930,7 @@ inside the package's `config/workflows.php` via
 | `DW_V2_LIMIT_MEMO_SIZE_BYTES` | `262144` | Package-level memo byte ceiling. |
 | `DW_V2_LIMIT_SEARCH_ATTRIBUTE_SIZE_BYTES` | `40960` | Package-level search-attribute byte ceiling. |
 | `DW_V2_LIMIT_HISTORY_TRANSACTION_SIZE` | `5000` | Package-level history-transaction event ceiling. |
+| `DW_V2_LIMIT_HISTORY_TRANSACTION_BYTES` | `16777216` | Package-level history-transaction byte ceiling, also applied to Waterline v1 projection imports. Zero disables this dimension. |
 | `DW_V2_LIMIT_WARNING_THRESHOLD_PERCENT` | `80` | Percent of a structural limit at which the package warns. |
 | `DW_V2_TASK_DISPATCH_MODE` | `queue` | Package-level workflow-task dispatch mode. Usually overridden by the server via `DW_TASK_DISPATCH_MODE`. |
 | `DW_V2_MATCHING_ROLE_QUEUE_WAKE` | `true` | Whether queue workers run the in-worker matching-role wake on every Looping event. In the supplied Compose stacks, `false` also skips the scheduler's unscoped repair pass. Set it to `false` only when a dedicated `php artisan workflow:v2:repair-pass --loop` daemon owns the sweep. |