Fence history import targets, messages and lineage

docs/workflow/plan.md

@@ -636,6 +636,12 @@ do not introduce new durable truth.
   version, payload codec, or compatibility marker the server does
   not declare support for, and `--dry-run` is the supported
   rehearsal mode for cutover audits.
+- Imported lineage resolves through edges whose run and instance endpoints
+  exist in the target namespace. Missing endpoints stay in the imported
+  history without attaching a later unrelated run. Importing the related
+  bundle can materialize the edge. PostgreSQL write imports must run outside
+  a caller transaction so the importer can own and retry its serializable
+  transaction. Preview imports do not write or change transaction isolation.
 - Cross-database import/export is the same `workflow:v2:history-export`
   / `workflow:v2:history-import` round trip with the language-neutral
   `avro` codec; the importer normalizes legacy `workflow-serializer-y`

src/V2/Support/CancellationScopeDelivery.php

@@ -667,6 +667,9 @@ private static function readBoundary(WorkflowRun $run, string $scopeId, bool $pr
                 );
             }
         } catch (LogicException $error) {
+            if ($error->getMessage() === 'cancellation_scope_child_target_mismatch') {
+                throw $error;
+            }
             throw new LogicException('cancellation_scope_delivery_history_invalid', previous: $error);
         }
         return $event;

src/V2/Support/ChildRunHistory.php

@@ -313,14 +313,18 @@
 
         $childRun = $childRunId === null
             ? $link?->childRun
-            : self::loadRun($childRunId);
+            : self::loadRun($childRunId, $run);
+
+        if ($childRun instanceof WorkflowRun && ! ImportedLineageReferences::allows($run, $childRun)) {
+            $childRun = null;
+        }
 
         if ($childRun instanceof WorkflowRun && ($startedEvent !== null || $resolutionEvent !== null)) {
             return $childRun;
         }
 
         if (! $childRun instanceof WorkflowRun && $childInstanceId !== null) {
-            $childRun = self::loadCurrentRunForInstance($childInstanceId);
+            $childRun = self::loadCurrentRunForInstance($childInstanceId, $run);
         }
 
         return self::followContinuedRun($childRun);
@@ -481,7 +485,7 @@
         $payload = is_array($startedEvent?->payload) ? $startedEvent->payload : null;
         $parentRunId = self::stringValue($payload['parent_workflow_run_id'] ?? null);
         $parentSequence = self::intValue($payload['parent_sequence'] ?? null);
-        $parentRun = $parentRunId === null ? null : self::loadRun($parentRunId);
+        $parentRun = $parentRunId === null ? null : self::loadRun($parentRunId, $run);
 
         return self::childCallIdFromPayload($payload)
             ?? self::legacyChildCallIdFromStartedWorkflowPayload($payload)
@@ -581,7 +585,10 @@
         WorkflowHistoryEvent $resolutionEvent,
         ?WorkflowRun $childRun = null,
     ): mixed {
-        $childRun ??= self::loadRun(self::stringValue($resolutionEvent->payload['child_workflow_run_id'] ?? null));
+        $relatedId = self::stringValue($resolutionEvent->payload['child_workflow_run_id'] ?? null);
+        if ($childRun === null && $relatedId !== null) {
+            $childRun = self::loadRun($relatedId, $resolutionEvent->run);
+        }
         ExecutionTracing::linkChildOutcome($resolutionEvent, $childRun);
         $payload = $resolutionEvent->payload['output'] ?? null;
 
@@ -907,34 +914,36 @@
         return $childCall;
     }
 
-    private static function loadRun(?string $runId): ?WorkflowRun
+    private static function loadRun(?string $runId, ?WorkflowRun $origin): ?WorkflowRun
     {
-        if ($runId === null) {
+        if ($runId === null || $origin === null) {
             return null;
         }
 
-        /** @var WorkflowRun|null $run */
-        $run = ConfiguredV2Models::query('run_model', WorkflowRun::class)
-            ->with(['summary', 'instance', 'failures', 'historyEvents'])
-            ->find($runId);
-
-        return $run;
+        return ImportedLineageReferences::resolve(
+            $origin,
+            $runId,
+            ['summary', 'instance', 'failures', 'historyEvents']
+        );
     }
 
-    private static function loadCurrentRunForInstance(string $instanceId): ?WorkflowRun
+    private static function loadCurrentRunForInstance(string $instanceId, WorkflowRun $origin): ?WorkflowRun
     {
         /** @var \Workflow\V2\Models\WorkflowInstance|null $instance */
         $instance = ConfiguredV2Models::query('instance_model', \Workflow\V2\Models\WorkflowInstance::class)
+            ->where('namespace', $origin->namespace)
             ->find($instanceId);
 
         if ($instance === null) {
             return null;
         }
 
-        return CurrentRunResolver::forInstance($instance, ['summary', 'failures', 'historyEvents']);
+        $run = CurrentRunResolver::forInstance($instance, ['summary', 'failures', 'historyEvents']);
+
+        return $run instanceof WorkflowRun && ImportedLineageReferences::allows($origin, $run) ? $run : null;
     }
 
     private static function followContinuedRun(?WorkflowRun $childRun): ?WorkflowRun
     {
         $visited = [];
 
@@ -950,6 +959,10 @@
             if (! $currentRun instanceof WorkflowRun || $currentRun->id === $childRun->id || isset($visited[$childRun->id])) {
                 return $childRun;
             }
+            if ($currentRun->namespace !== $childRun->namespace
+                || $currentRun->workflow_instance_id !== $childRun->workflow_instance_id) {
+                return $childRun;
+            }
 
             $visited[$childRun->id] = true;
             $childRun = $currentRun;

src/V2/Support/DefaultWorkflowTaskBridge.php

@@ -5048,8 +5048,30 @@ private function applyContinueAsNew(
             ->get();
 
         $parentRunsToProject = [];
+        $trustedParentReference = null;
 
         foreach ($parentChildLinks as $parentChildLink) {
+            /** @var WorkflowRun|null $parentRun */
+            $parentRun = is_string($parentChildLink->parent_workflow_run_id)
+                && $parentChildLink->parent_workflow_run_id !== ''
+                ? ConfiguredV2Models::query('run_model', WorkflowRun::class)
+                    ->lockForUpdate()
+                    ->find($parentChildLink->parent_workflow_run_id)
+                : null;
+            $requiresImportedEdge = ImportedLineageReferences::imported($run)
+                || ($parentRun instanceof WorkflowRun && ImportedLineageReferences::imported($parentRun));
+            if ($requiresImportedEdge && (! $parentRun instanceof WorkflowRun
+                || ! is_int($parentChildLink->sequence)
+                || ! ImportedLineageReferences::allowsChild(
+                    $parentRun,
+                    $run,
+                    $parentChildLink->sequence,
+                    $childCallId,
+                    true,
+                ))) {
+                continue;
+            }
+
             /** @var WorkflowLink $continuedChildLink */
             $continuedChildLink = WorkflowLink::query()->create([
                 'link_type' => 'child_workflow',
@@ -5063,20 +5085,17 @@ private function applyContinueAsNew(
                 'parent_close_policy' => $parentChildLink->parent_close_policy,
             ]);
 
-            if (
-                ! is_string($parentChildLink->parent_workflow_run_id)
-                || $parentChildLink->parent_workflow_run_id === ''
-                || ! is_int($parentChildLink->sequence)
-            ) {
-                continue;
+            if (ImportedLineageReferences::imported($run) && $parentRun instanceof WorkflowRun
+                && is_int($parentChildLink->sequence) && $trustedParentReference === null) {
+                $trustedParentReference = [
+                    'parent_workflow_instance_id' => $parentRun->workflow_instance_id,
+                    'parent_workflow_run_id' => $parentRun->id,
+                    'parent_sequence' => $parentChildLink->sequence,
+                ];
             }
 
-            /** @var WorkflowRun|null $parentRun */
-            $parentRun = ConfiguredV2Models::query('run_model', WorkflowRun::class)
-                ->lockForUpdate()
-                ->find($parentChildLink->parent_workflow_run_id);
-
-            if ($parentRun === null || $parentRun->status->isTerminal()) {
+            if (! $parentRun instanceof WorkflowRun || ! is_int($parentChildLink->sequence)
+                || $parentRun->status->isTerminal()) {
                 continue;
             }
 
@@ -5118,7 +5137,9 @@ private function applyContinueAsNew(
             ], $parallelMetadata), static fn (mixed $value): bool => $value !== null));
         }
 
-        $parentReference = ChildRunHistory::parentReferenceForRun($run);
+        $parentReference = ImportedLineageReferences::imported($run)
+            ? $trustedParentReference
+            : ChildRunHistory::parentReferenceForRun($run);
 
         WorkflowHistoryEvent::record($continuedRun, HistoryEventType::WorkflowStarted, [
             'workflow_class' => $workflowType,

src/V2/Support/EmbeddedV2HistoryImport.php

@@ -146,7 +146,12 @@
             ];
         }
 
-        $existing = self::existingRunOutcome($workflow['run_id'], self::stringValue($bundle['dedupe_key'] ?? null));
+        $namespace = self::stringValue($workflow['namespace'] ?? null);
+        $existing = self::existingRunOutcome(
+            $workflow['run_id'],
+            self::stringValue($bundle['dedupe_key'] ?? null),
+            $namespace
+        );
 
         if ($existing['status'] === 'already_imported') {
             return [
@@ -169,6 +174,20 @@
             ];
         }
 
+        $instance = WorkflowInstance::query()->find($workflow['instance_id']);
+        if ($instance instanceof WorkflowInstance && $instance->namespace !== $namespace) {
+            return [
+                ...$base,
+                'status' => 'rejected',
+                'eligibility' => self::withEligibilityError(
+                    $eligibility,
+                    'target.instance_namespace_conflict',
+                    'A workflow instance with this instance_id already exists outside the target namespace.',
+                ),
+                'rows' => self::emptyRows(),
+            ];
+        }
+
         if ($dryRun) {
             return [
                 ...$base,
@@ -177,13 +196,33 @@
             ];
         }
 
+        if (DB::connection()->getDriverName() === 'pgsql' && DB::transactionLevel() !== 0) {
+            return [
+                ...$base,
+                'status' => 'rejected',
+                'eligibility' => self::withEligibilityError(
+                    $eligibility,
+                    'target.transaction_in_progress',
+                    'PostgreSQL history import must run outside an existing transaction so its serializable transaction can be retried safely.',
+                ),
+                'rows' => self::emptyRows(),
+            ];
+        }
+
         $importId = self::stringValue($options['import_id'] ?? null) ?? (string) Str::ulid();
         $importedAt = now();
 
         try {
             $rows = DB::transaction(static function () use ($bundle, $workflow, $importId, $importedAt): array {
+                if (DB::connection()->getDriverName() === 'pgsql') {
+                    // Row locks cannot lock a missing endpoint on PostgreSQL.
+                    // Serializable predicate checks prevent reciprocal imports
+                    // from both committing without seeing the other's run.
+                    DB::statement('SET TRANSACTION ISOLATION LEVEL SERIALIZABLE');
+                }
+
                 return self::writeBundle($bundle, $workflow, $importId, $importedAt);
-            });
+            }, 3);
         } catch (Throwable $throwable) {
             return [
                 ...$base,
@@ -217,6 +256,10 @@
         string $importId,
         CarbonInterface $importedAt,
     ): array {
+        $lineageErrors = EmbeddedV2ImportLineage::errors($bundle, $workflow, true);
+        if ($lineageErrors !== []) {
+            throw new InvalidArgumentException($lineageErrors[0]['message']);
+        }
         $rows = self::emptyRows();
         $namespace = self::stringValue($workflow['namespace'] ?? null);
         $runStatus = RunStatus::from($workflow['status']);
@@ -240,6 +283,7 @@
 
         /** @var WorkflowInstance|null $instance */
         $instance = WorkflowInstance::query()
+            ->where('namespace', $namespace)
             ->lockForUpdate()
             ->find($instanceId);
 
@@ -319,7 +363,7 @@
         $rows['activity_attempts'] += self::writeActivityAttempts($bundle, $runId);
         $rows['workflow_run_timers'] += self::writeTimers($bundle, $runId);
         $rows['workflow_failures'] += self::writeFailures($bundle, $runId);
-        $rows['workflow_links'] += self::writeLinks($bundle, $runId, $instanceId);
+        $rows['workflow_links'] += self::writeLinks($bundle, $runId, $instanceId, $namespace);
         $rows['workflow_memos'] += self::writeMemos($workflow, $runId, $instanceId, $historySequence);
         $rows['workflow_search_attributes'] += self::writeSearchAttributes(
             $workflow,
@@ -707,7 +751,7 @@
         return $count;
     }
 
-    private static function writeLinks(array $bundle, string $runId, string $instanceId): int
+    private static function writeLinks(array $bundle, string $runId, string $instanceId, ?string $namespace): int
     {
         $links = self::arrayValue($bundle['links'] ?? null);
         $count = 0;
@@ -719,8 +763,12 @@
             if ($parentInstance === null || $parentRun === null) {
                 continue;
             }
+            $related = EmbeddedV2ImportLineage::existingEndpoint($namespace, $parentRun, $parentInstance, $runId);
+            if ($related === null) {
+                continue;
+            }
 
-            $count += self::insertLink([
+            $inserted = self::insertLink([
                 'id' => self::ulidValue($link['id'] ?? null) ?? (string) Str::ulid(),
                 'link_type' => self::stringValue($link['type'] ?? null) ?? 'parent',
                 'sequence' => self::intValue($link['sequence'] ?? null),
@@ -731,7 +779,11 @@
                 'is_primary_parent' => (bool) ($link['is_primary_parent'] ?? false),
                 'created_at' => self::timestamp($link['created_at'] ?? null) ?? now(),
                 'updated_at' => self::timestamp($link['created_at'] ?? null) ?? now(),
-            ]) ? 1 : 0;
+            ]);
+            if ($inserted) {
+                $count++;
+                RunSummaryProjector::project($related->fresh() ?? $related);
+            }
         }
 
         foreach (self::listValue($links['children'] ?? null) as $link) {
@@ -741,8 +793,12 @@
             if ($childInstance === null || $childRun === null) {
                 continue;
             }
+            $related = EmbeddedV2ImportLineage::existingEndpoint($namespace, $childRun, $childInstance, $runId);
+            if ($related === null) {
+                continue;
+            }
 
-            $count += self::insertLink([
+            $inserted = self::insertLink([
                 'id' => self::ulidValue($link['id'] ?? null) ?? (string) Str::ulid(),
                 'link_type' => self::stringValue($link['type'] ?? null) ?? 'child',
                 'sequence' => self::intValue($link['sequence'] ?? null),
@@ -753,7 +809,11 @@
                 'is_primary_parent' => (bool) ($link['is_primary_parent'] ?? false),
                 'created_at' => self::timestamp($link['created_at'] ?? null) ?? now(),
                 'updated_at' => self::timestamp($link['created_at'] ?? null) ?? now(),
-            ]) ? 1 : 0;
+            ]);
+            if ($inserted) {
+                $count++;
+                RunSummaryProjector::project($related->fresh() ?? $related);
+            }
         }
 
         return $count;
@@ -978,6 +1038,9 @@
             self::rejectOpenLeases($bundle, $errors);
         }
 
+        $errors = [...$errors, ...EmbeddedV2ImportMessageReferences::errors($bundle)];
+        $errors = [...$errors, ...EmbeddedV2ImportLineage::errors($bundle, $workflow)];
+
         return [
             'eligible' => $errors === [],
             'errors' => $errors,
@@ -1184,7 +1247,7 @@
     /**
      * @return array{status: string}
      */
-    private static function existingRunOutcome(?string $runId, ?string $dedupeKey): array
+    private static function existingRunOutcome(?string $runId, ?string $dedupeKey, ?string $namespace): array
     {
         if ($runId === null) {
             return [
@@ -1202,7 +1265,8 @@
         }
 
         if (
-            $run->import_source === EmbeddedV2ImportContract::IMPORT_SOURCE
+            $run->namespace === $namespace
+            && $run->import_source === EmbeddedV2ImportContract::IMPORT_SOURCE
             && is_string($run->import_dedupe_key)
             && $run->import_dedupe_key === $dedupeKey
         ) {

src/V2/Support/EmbeddedV2ImportContract.php

@@ -85,12 +85,13 @@ public static function manifest(): array
                 'activity_tasks' => 'pending activity executions and ready activity tasks remain claimable; running attempts are not importable',
                 'waits' => 'wait projections are rebuilt from typed history plus imported task/activity/timer rows',
                 'timers' => 'pending timers are copied with their fire_at timestamp and are visible to timer repair/recovery',
-                'lineage' => 'parent and child links from the bundle are copied as diagnostic lineage edges when identifiers are present',
+                'lineage' => 'parent and child edges are copied only when both endpoint run/instance pairs exist in the target namespace; missing endpoints remain in history without resolving later unrelated runs, and the counterpart bundle can materialize the edge',
             ],
             'failure_and_rollback' => [
                 'transaction' => 'all durable rows and projections are written in one database transaction',
+                'postgresql_transaction' => 'write imports own a serializable transaction and reject an existing caller transaction; serialization conflicts retry the complete transaction up to three attempts without changing session defaults',
                 'partial_or_interrupted_import' => 'no imported rows are committed when the transaction rolls back',
-                'retry_same_bundle' => 'returns already_imported when run_id and import_dedupe_key match',
+                'retry_same_bundle' => 'returns already_imported when namespace, import source, run_id and import_dedupe_key match',
                 'conflicting_existing_run' => 'rejected before writing any rows',
             ],
             'visibility_and_audit' => [