"""P8 inventory, backfill, shadow validation, and per-Workspace atomic cutover. Inputs are strict operator-produced snapshots and remediation evidence. This module never discovers or matches legacy product rows: opaque KnowledgeFS Space IDs are registered directly into the independent control plane. Unknown access, unresolved subjects, and unresolved cutover quarantine remain fail-closed and auditable. """ from __future__ import annotations import hashlib import json from collections.abc import Callable, Iterable from datetime import UTC, datetime from typing import Literal, NamedTuple, cast from uuid import UUID import sqlalchemy as sa from pydantic import BaseModel, ConfigDict, Field, JsonValue, ValidationError, field_validator, model_validator from sqlalchemy.orm import Session, sessionmaker from libs.datetime_utils import naive_utc_now from models.knowledge_fs import ( KnowledgeFSApiCredential, KnowledgeFSApiCredentialStatus, KnowledgeFSAuthorizationRevision, KnowledgeFSControlSpace, KnowledgeFSControlSpacePermission, KnowledgeFSControlSpacePermissionRole, KnowledgeFSControlSpaceState, KnowledgeFSControlSpaceVisibility, KnowledgeFSExternalAccessPolicy, ) from models.knowledge_fs_cutover import ( KnowledgeFSCutoverRevisionWatermark, KnowledgeFSCutoverSmokeResults, KnowledgeFSMigrationIssue, KnowledgeFSMigrationIssueKind, KnowledgeFSMigrationIssueStatus, KnowledgeFSMigrationQuarantine, KnowledgeFSMigrationQuarantineDisposition, KnowledgeFSMigrationQuarantineKind, KnowledgeFSShadowAuthorizationDecision, KnowledgeFSShadowAuthorizationDiff, KnowledgeFSShadowAuthorizationObservation, KnowledgeFSWorkspaceCutoverLedger, KnowledgeFSWorkspaceCutoverPhase, knowledge_fs_cutover_smoke_results_passed, ) from repositories.knowledge_fs_cutover_repository import ( KnowledgeFSCutoverCASUpdate, KnowledgeFSQuarantineCASUpdate, KnowledgeFSShadowDiffCASUpdate, ) from repositories.sqlalchemy_knowledge_fs_cutover_repository import SQLAlchemyKnowledgeFSCutoverRepository from services.knowledge_fs.lifecycle_port import ( KnowledgeFSDifyIntegrationActivationAck, KnowledgeFSDifyIntegrationActivationRequest, KnowledgeFSDifyIntegrationFreezeAck, KnowledgeFSDifyIntegrationFreezeRequest, KnowledgeFSLifecycleRemoteError, KnowledgeFSLifecycleRemotePort, ) class StrictInput(BaseModel): model_config = ConfigDict(extra="forbid", frozen=True) class CutoverRevisionWatermarkInput(StrictInput): membership_epoch: int = Field(ge=0) space_acl_epoch: int = Field(ge=0) external_access_epoch: int = Field(ge=0) content_policy_revision: int = Field(ge=0) def to_record(self) -> KnowledgeFSCutoverRevisionWatermark: return cast(KnowledgeFSCutoverRevisionWatermark, self.model_dump()) class LegacyPermissionInventoryInput(StrictInput): subject_id: str = Field(min_length=1, max_length=255) account_id: UUID | None = None role: Literal["owner", "editor", "viewer"] class LegacyTaskInventoryInput(StrictInput): task_id: str = Field(min_length=1, max_length=255) subject_id: str | None = Field(default=None, min_length=1, max_length=255) account_id: UUID | None = None state: Literal["queued", "running", "paused", "completed", "failed", "canceled"] expires_at: datetime | None = None class LegacyApiKeyInventoryInput(StrictInput): key_id: str = Field(min_length=1, max_length=255) prefix: str = Field(min_length=1, max_length=32) last4: str = Field(min_length=4, max_length=4) class LegacySpaceInventoryInput(StrictInput): knowledge_space_id: UUID knowledge_space_revision: int = Field(ge=0) provisioning_key: str = Field(min_length=1, max_length=255) owner_subject_id: str = Field(min_length=1, max_length=255) owner_account_id: UUID | None = None visibility: Literal["only_me", "all_members", "partial_members", "unknown"] external_access_enabled: bool | None = None permissions: list[LegacyPermissionInventoryInput] = Field(default_factory=list) legacy_api_keys: list[LegacyApiKeyInventoryInput] = Field(default_factory=list) tasks: list[LegacyTaskInventoryInput] = Field(default_factory=list) orphan_resource_ids: list[str] = Field(default_factory=list) class WorkspaceInventoryInput(StrictInput): tenant_id: UUID source_revision_watermark: CutoverRevisionWatermarkInput task_watermark: int = Field(ge=0) spaces: list[LegacySpaceInventoryInput] class ShadowAuthorizationObservationInput(StrictInput): schema_version: Literal["knowledge-fs-p8-shadow-observation/v1"] tenant_id: UUID diff_key: str = Field(min_length=1, max_length=255) control_space_id: UUID | None = None principal: str = Field(min_length=1, max_length=255) legacy_allowed: bool | None dify_allowed: bool reason: str = Field(min_length=1, max_length=4096) observed_revision: CutoverRevisionWatermarkInput observed_at: datetime producer: str = Field(min_length=1, max_length=255) @field_validator("observed_at") @classmethod def require_timezone(cls, value: datetime) -> datetime: if value.tzinfo is None or value.utcoffset() is None: raise ValueError("shadow observation timestamps require an explicit timezone") return value class ShadowCompletionInput(StrictInput): schema_version: Literal["knowledge-fs-p8-shadow-completion/v1"] tenant_id: UUID expected_cas_version: int = Field(ge=0) producer: str = Field(min_length=1, max_length=255) completed_by_operator: str = Field(min_length=1, max_length=255) completed_by_account_id: UUID completed_at: datetime traffic_zero: bool window_started_at: datetime | None = None window_ended_at: datetime | None = None traffic_zero_evidence: dict[str, JsonValue] | None = Field(default=None, min_length=1, max_length=32) @field_validator("completed_at", "window_started_at", "window_ended_at") @classmethod def require_timezone(cls, value: datetime | None) -> datetime | None: if value is not None and (value.tzinfo is None or value.utcoffset() is None): raise ValueError("shadow completion timestamps require an explicit timezone") return value @model_validator(mode="after") def require_window_or_traffic_zero_evidence(self) -> ShadowCompletionInput: if self.traffic_zero: if self.traffic_zero_evidence is None: raise ValueError("traffic-zero completion requires explicit evidence") if self.window_started_at is not None or self.window_ended_at is not None: raise ValueError("traffic-zero completion cannot also declare an observation window") elif self.window_started_at is None or self.window_ended_at is None or self.traffic_zero_evidence is not None: raise ValueError("nonzero shadow completion requires a window and no traffic-zero evidence") return self class FinalDeltaInput(StrictInput): tenant_id: UUID expected_cas_version: int = Field(ge=0) final_revision_watermark: CutoverRevisionWatermarkInput applied_revision_watermark: CutoverRevisionWatermarkInput final_task_watermark: int = Field(ge=0) applied_task_watermark: int = Field(ge=0) class GeneralQuarantineResolutionEvidence(StrictInput): schema_version: Literal["knowledge-fs-p8-general-resolution/v1"] reference: str = Field(min_length=1, max_length=2048) class LegacyCredentialRotationEvidence(StrictInput): schema_version: Literal["knowledge-fs-p8-credential-rotation/v1"] legacy_key_id: str = Field(min_length=1, max_length=255) knowledge_space_id: UUID control_space_id: UUID dify_credential_id: UUID dify_credential_revision: int = Field(ge=0) legacy_revoked_at: datetime verification_reference: str = Field(min_length=1, max_length=2048) plaintext_migrated: Literal[False] @field_validator("legacy_revoked_at") @classmethod def require_timezone(cls, value: datetime) -> datetime: if value.tzinfo is None or value.utcoffset() is None: raise ValueError("credential rotation timestamp requires an explicit timezone") return value class LegacyTaskResolutionEvidence(StrictInput): schema_version: Literal["knowledge-fs-p8-task-resolution/v1"] task_id: str = Field(min_length=1, max_length=255) action: Literal["migrate", "wait", "cancel", "isolate"] resulting_state: Literal["migrated", "completed", "failed", "canceled", "isolated"] final_task_watermark: int = Field(ge=0) verification_reference: str = Field(min_length=1, max_length=2048) class QuarantineResolutionInput(StrictInput): """Operator-attested remediation evidence for one tenant-scoped quarantine row.""" schema_version: Literal["knowledge-fs-p8-quarantine-resolution/v1"] tenant_id: UUID source_kind: KnowledgeFSMigrationQuarantineKind source_id: str = Field(min_length=1, max_length=255) expected_row_version: int = Field(ge=0) resolved_by_operator: str = Field(min_length=1, max_length=255) resolved_by_account_id: UUID evidence: GeneralQuarantineResolutionEvidence | LegacyCredentialRotationEvidence | LegacyTaskResolutionEvidence resolved_at: datetime @field_validator("resolved_at") @classmethod def require_timezone(cls, value: datetime) -> datetime: if value.tzinfo is None or value.utcoffset() is None: raise ValueError("quarantine resolution timestamps require an explicit timezone") return value @model_validator(mode="after") def require_source_specific_evidence(self) -> QuarantineResolutionInput: expected_type: type[StrictInput] if self.source_kind is KnowledgeFSMigrationQuarantineKind.LEGACY_API_KEY: expected_type = LegacyCredentialRotationEvidence elif self.source_kind is KnowledgeFSMigrationQuarantineKind.TASK: expected_type = LegacyTaskResolutionEvidence else: expected_type = GeneralQuarantineResolutionEvidence if not isinstance(self.evidence, expected_type): raise ValueError(f"{self.source_kind.value} requires its source-specific evidence schema") return self class CutoverSmokeChecksInput(StrictInput): authorization: bool list_spaces: bool create_space: bool query: bool upload: bool stream: bool deletion: bool class CutoverSmokeEvidenceReferencesInput(StrictInput): authorization: str = Field(min_length=1, max_length=2048) list_spaces: str = Field(min_length=1, max_length=2048) create_space: str = Field(min_length=1, max_length=2048) query: str = Field(min_length=1, max_length=2048) upload: str = Field(min_length=1, max_length=2048) stream: str = Field(min_length=1, max_length=2048) deletion: str = Field(min_length=1, max_length=2048) class CutoverSmokeResultsInput(StrictInput): schema_version: Literal["knowledge-fs-p8-cutover-smoke/v1"] tenant_id: UUID environment: Literal["production"] operator: str = Field(min_length=1, max_length=255) operator_account_id: UUID observed_at: datetime checks: CutoverSmokeChecksInput evidence_references: CutoverSmokeEvidenceReferencesInput @field_validator("observed_at") @classmethod def require_timezone(cls, value: datetime) -> datetime: if value.tzinfo is None or value.utcoffset() is None: raise ValueError("smoke evidence timestamp requires an explicit timezone") return value def to_record(self) -> KnowledgeFSCutoverSmokeResults: return cast(KnowledgeFSCutoverSmokeResults, self.model_dump(mode="json")) @property def passed(self) -> bool: return all(self.checks.model_dump().values()) class LegacyDependencyInput(StrictInput): dependency_key: str = Field(min_length=1, max_length=255) kind: Literal["permission_snapshot", "foreign_key"] table_name: str = Field(min_length=1, max_length=255) column_name: str = Field(min_length=1, max_length=255) constraint_name: str | None = Field(default=None, min_length=1, max_length=255) active_rows: int = Field(ge=0) migrated_rows: int = Field(ge=0) class CutoverInventoryReport(NamedTuple): tenant_id: str spaces: int permissions: int unresolved_subjects: int unknown_external_access: int legacy_keys_requiring_rotation: int tasks_by_disposition: dict[str, int] orphan_resources: int applied: bool replayed: bool class CutoverBackfillReport(NamedTuple): tenant_id: str control_spaces_registered: int permissions_granted: int quarantined: int open_issues: int phase: str applied: bool class ShadowAuthorizationReport(NamedTuple): tenant_id: str matches: int tightened: int expanded: int unknown: int open_diffs: int applied: bool recorded: int replayed: int class ShadowCompletionReport(NamedTuple): tenant_id: str observation_count: int traffic_zero: bool evidence_digest: str latest_observed_revision: KnowledgeFSCutoverRevisionWatermark | None applied: bool replayed: bool class LegacyDependencyDashboard(NamedTuple): tenant_id: str ready: bool permission_snapshot_dependencies: int foreign_key_dependencies: int active_rows: int dependencies: tuple[dict[str, object], ...] applied: bool class QuarantineResolutionReport(NamedTuple): tenant_id: str source_kind: str source_id: str disposition: str row_version: int applied: bool replayed: bool class KnowledgeFSCutoverError(RuntimeError): """Base error for an operator cutover request rejected before mutation.""" class KnowledgeFSCutoverNotFoundError(KnowledgeFSCutoverError): pass class KnowledgeFSCutoverConflictError(KnowledgeFSCutoverError): pass class KnowledgeFSCutoverGateBlockedError(KnowledgeFSCutoverError): pass _VISIBILITY_MAP = { "only_me": KnowledgeFSControlSpaceVisibility.ONLY_ME, "all_members": KnowledgeFSControlSpaceVisibility.ALL_TEAM_MEMBERS, "partial_members": KnowledgeFSControlSpaceVisibility.PARTIAL_MEMBERS, } _CUTOVER_QUARANTINE_KINDS = ( KnowledgeFSMigrationQuarantineKind.CONTROL_SPACE, KnowledgeFSMigrationQuarantineKind.TASK, KnowledgeFSMigrationQuarantineKind.LEGACY_API_KEY, KnowledgeFSMigrationQuarantineKind.ORPHAN_RESOURCE, ) _DEFAULT_TRUSTED_SHADOW_PRODUCERS = frozenset({"dify-shadow-authorizer"}) _DEFAULT_TRUSTED_SHADOW_OPERATORS = frozenset({"knowledge-fs-cutover"}) class KnowledgeFSWorkspaceCutoverService: """Coordinate one Workspace through the strict P8 phase sequence.""" _session_maker: sessionmaker[Session] _clock: Callable[[], datetime] _trusted_shadow_producers: frozenset[str] _trusted_shadow_operators: frozenset[str] _remote_factory: Callable[[], KnowledgeFSLifecycleRemotePort] | None def __init__( self, session_maker: sessionmaker[Session], clock: Callable[[], datetime] = naive_utc_now, *, trusted_shadow_producers: frozenset[str] = _DEFAULT_TRUSTED_SHADOW_PRODUCERS, trusted_shadow_operators: frozenset[str] = _DEFAULT_TRUSTED_SHADOW_OPERATORS, remote_factory: Callable[[], KnowledgeFSLifecycleRemotePort] | None = None, ): self._session_maker = session_maker self._clock = clock self._trusted_shadow_producers = trusted_shadow_producers self._trusted_shadow_operators = trusted_shadow_operators self._remote_factory = remote_factory def inventory(self, payload: WorkspaceInventoryInput, *, apply: bool) -> CutoverInventoryReport: """Validate a read-only inventory and optionally create the initial CAS ledger.""" tenant_id = str(payload.tenant_id) self._assert_unique_inventory(payload.spaces) dispositions = self._task_disposition_counts(payload.spaces) report = CutoverInventoryReport( tenant_id=tenant_id, spaces=len(payload.spaces), permissions=sum(len(space.permissions) for space in payload.spaces), unresolved_subjects=sum(space.owner_account_id is None for space in payload.spaces) + sum(permission.account_id is None for space in payload.spaces for permission in space.permissions), unknown_external_access=sum(space.external_access_enabled is None for space in payload.spaces), legacy_keys_requiring_rotation=sum(len(space.legacy_api_keys) for space in payload.spaces), tasks_by_disposition=dispositions, orphan_resources=sum(len(space.orphan_resource_ids) for space in payload.spaces), applied=apply, replayed=False, ) if not apply: return report source_watermark = payload.source_revision_watermark.to_record() with self._session_maker.begin() as session: repository = SQLAlchemyKnowledgeFSCutoverRepository(session) current = repository.get_ledger(tenant_id=tenant_id) if current is not None: if ( current.source_revision_watermark != source_watermark or current.source_task_watermark != payload.task_watermark ): raise KnowledgeFSCutoverConflictError("Workspace inventory watermarks conflict with the ledger") return report._replace(replayed=True) repository.add_ledger( KnowledgeFSWorkspaceCutoverLedger( tenant_id=tenant_id, source_revision_watermark=source_watermark, applied_revision_watermark=source_watermark, source_task_watermark=payload.task_watermark, applied_task_watermark=payload.task_watermark, ) ) return report def backfill(self, payload: WorkspaceInventoryInput, *, apply: bool) -> CutoverBackfillReport: """Backfill only independent control-plane rows; unresolved inputs stay quarantined.""" tenant_id = str(payload.tenant_id) self._assert_unique_inventory(payload.spaces) if not apply: return self._project_backfill(payload) registered = 0 granted = 0 quarantined = 0 with self._session_maker.begin() as session: repository = SQLAlchemyKnowledgeFSCutoverRepository(session) ledger = self._require_ledger(repository, tenant_id) if ledger.phase not in { KnowledgeFSWorkspaceCutoverPhase.INVENTORY, KnowledgeFSWorkspaceCutoverPhase.BACKFILL, }: raise KnowledgeFSCutoverConflictError("Backfill is closed after shadow validation starts") if ( ledger.source_revision_watermark != payload.source_revision_watermark.to_record() or ledger.source_task_watermark != payload.task_watermark ): raise KnowledgeFSCutoverConflictError("Backfill input does not match the inventoried watermarks") for space in payload.spaces: result = self._backfill_space( session, repository, ledger, space, payload.source_revision_watermark.to_record(), ) registered += result[0] granted += result[1] quarantined += result[2] session.flush() open_issues = repository.count_open_issues(tenant_id=tenant_id, ledger_id=ledger.id) if ledger.phase is KnowledgeFSWorkspaceCutoverPhase.INVENTORY and open_issues == 0: self._cas( repository, KnowledgeFSCutoverCASUpdate( tenant_id=tenant_id, expected_phase=ledger.phase, expected_cas_version=ledger.cas_version, new_phase=KnowledgeFSWorkspaceCutoverPhase.BACKFILL, ), ) phase = KnowledgeFSWorkspaceCutoverPhase.BACKFILL.value else: phase = ledger.phase.value return CutoverBackfillReport(tenant_id, registered, granted, quarantined, open_issues, phase, True) def begin_shadow( self, *, tenant_id: str, expected_cas_version: int, started_at: datetime | None = None, ) -> KnowledgeFSWorkspaceCutoverLedger: shadow_started_at = _naive_utc(started_at or self._clock()) if shadow_started_at > _naive_utc(self._clock()): raise KnowledgeFSCutoverConflictError("Shadow start timestamp cannot be in the future") with self._session_maker.begin() as session: repository = SQLAlchemyKnowledgeFSCutoverRepository(session) ledger = self._require_ledger(repository, tenant_id) self._assert_phase(ledger, KnowledgeFSWorkspaceCutoverPhase.BACKFILL, expected_cas_version) self._assert_no_open_gates(repository, ledger) self._cas( repository, KnowledgeFSCutoverCASUpdate( tenant_id=tenant_id, expected_phase=ledger.phase, expected_cas_version=ledger.cas_version, new_phase=KnowledgeFSWorkspaceCutoverPhase.SHADOW, shadow_started_at=shadow_started_at, ), ) return self._require_ledger(repository, tenant_id) def record_shadow_report( self, observations: Iterable[ShadowAuthorizationObservationInput], *, apply: bool, ) -> ShadowAuthorizationReport: items = tuple(observations) if not items: raise KnowledgeFSCutoverConflictError("Shadow report requires at least one observation") tenant_ids = {str(item.tenant_id) for item in items} if len(tenant_ids) != 1: raise KnowledgeFSCutoverConflictError("Shadow report cannot mix Workspaces") tenant_id = tenant_ids.pop() if len({item.producer for item in items}) != 1: raise KnowledgeFSCutoverConflictError("Shadow report cannot mix producers") decisions = [self._shadow_decision(item) for item in items] counts = {decision: decisions.count(decision) for decision in KnowledgeFSShadowAuthorizationDecision} replayed = 0 with self._session_maker.begin() as session: repository = SQLAlchemyKnowledgeFSCutoverRepository(session) ledger = self._require_ledger(repository, tenant_id) if ledger.phase is not KnowledgeFSWorkspaceCutoverPhase.SHADOW: raise KnowledgeFSCutoverConflictError("Shadow observations require the shadow phase") for item, decision in zip(items, decisions, strict=True): self._validate_shadow_observation(ledger, item) replayed += self._record_shadow_observation( repository, ledger, item, decision, apply=apply, ) if apply: session.flush() open_diffs = repository.count_unapproved_shadow_diffs(tenant_id=tenant_id, ledger_id=ledger.id) else: open_diffs = sum(decision is not KnowledgeFSShadowAuthorizationDecision.MATCH for decision in decisions) return ShadowAuthorizationReport( tenant_id, counts[KnowledgeFSShadowAuthorizationDecision.MATCH], counts[KnowledgeFSShadowAuthorizationDecision.TIGHTENED], counts[KnowledgeFSShadowAuthorizationDecision.EXPANDED], counts[KnowledgeFSShadowAuthorizationDecision.UNKNOWN], open_diffs, apply, len(items) - replayed, replayed, ) def complete_shadow(self, payload: ShadowCompletionInput, *, apply: bool) -> ShadowCompletionReport: """Close a trusted shadow window and persist an immutable digest of its evidence.""" tenant_id = str(payload.tenant_id) completed_at = _naive_utc(payload.completed_at) now = _naive_utc(self._clock()) if payload.producer not in self._trusted_shadow_producers: raise KnowledgeFSCutoverGateBlockedError("Shadow completion producer is not trusted") if payload.completed_by_operator not in self._trusted_shadow_operators: raise KnowledgeFSCutoverGateBlockedError("Shadow completion operator is not trusted") if completed_at > now: raise KnowledgeFSCutoverConflictError("Shadow completion timestamp cannot be in the future") with self._session_maker.begin() as session: repository = SQLAlchemyKnowledgeFSCutoverRepository(session) ledger = self._require_ledger(repository, tenant_id) if ledger.phase is not KnowledgeFSWorkspaceCutoverPhase.SHADOW: raise KnowledgeFSCutoverConflictError("Shadow completion requires the shadow phase") if ledger.shadow_started_at is None: raise KnowledgeFSCutoverGateBlockedError("Shadow start evidence is missing") if completed_at < ledger.shadow_started_at: raise KnowledgeFSCutoverConflictError("Shadow completion predates shadow start") observations = repository.list_shadow_observations(tenant_id=tenant_id, ledger_id=ledger.id) latest_revision: KnowledgeFSCutoverRevisionWatermark | None if payload.traffic_zero: if observations: raise KnowledgeFSCutoverConflictError( "Traffic-zero completion conflicts with persisted observations" ) latest_revision = None evidence_digest = _shadow_traffic_zero_digest(payload) else: if not observations: raise KnowledgeFSCutoverGateBlockedError( "Shadow completion requires at least one observation or traffic-zero evidence" ) if any(item.producer != payload.producer for item in observations): raise KnowledgeFSCutoverGateBlockedError( "Shadow observations were not produced by the completing trusted producer" ) window_started_at = _naive_utc(cast(datetime, payload.window_started_at)) window_ended_at = _naive_utc(cast(datetime, payload.window_ended_at)) if window_started_at < ledger.shadow_started_at or window_ended_at <= window_started_at: raise KnowledgeFSCutoverConflictError("Shadow observation window is invalid") if window_ended_at > completed_at: raise KnowledgeFSCutoverConflictError("Shadow observation window ends after completion") if any( item.observed_at < window_started_at or item.observed_at > window_ended_at for item in observations ): raise KnowledgeFSCutoverGateBlockedError( "Persisted shadow observations fall outside the completed window" ) latest_revision = _latest_shadow_revision(observations) required_revision = ledger.final_revision_watermark or ledger.source_revision_watermark if not _watermark_at_least(latest_revision, required_revision): raise KnowledgeFSCutoverGateBlockedError( "Shadow latest observed revision is below the applicable cutover watermark" ) evidence_digest = _shadow_observation_set_digest(observations) report = ShadowCompletionReport( tenant_id, len(observations), payload.traffic_zero, evidence_digest, latest_revision, apply, ledger.shadow_completed_at is not None, ) if ledger.shadow_completed_at is not None: self._assert_shadow_completion_replay(ledger, payload, report) return report if ledger.cas_version != payload.expected_cas_version: raise KnowledgeFSCutoverConflictError("Cutover ledger version changed") if not apply: return report self._cas( repository, KnowledgeFSCutoverCASUpdate( tenant_id=tenant_id, expected_phase=ledger.phase, expected_cas_version=ledger.cas_version, new_phase=ledger.phase, shadow_completed_at=completed_at, shadow_evidence_digest=evidence_digest, shadow_observation_count=len(observations), shadow_window_started_at=( _naive_utc(payload.window_started_at) if payload.window_started_at is not None else None ), shadow_window_ended_at=( _naive_utc(payload.window_ended_at) if payload.window_ended_at is not None else None ), shadow_traffic_zero=payload.traffic_zero, shadow_traffic_zero_evidence=( cast(dict[str, object], payload.traffic_zero_evidence) if payload.traffic_zero_evidence is not None else None ), shadow_latest_observed_revision=latest_revision, shadow_producer=payload.producer, shadow_completed_by_operator=payload.completed_by_operator, shadow_completed_by_account_id=str(payload.completed_by_account_id), ), ) return report def approve_shadow_diff( self, *, tenant_id: str, diff_key: str, account_id: str, approved_at: datetime, ) -> None: with self._session_maker.begin() as session: repository = SQLAlchemyKnowledgeFSCutoverRepository(session) ledger = self._require_ledger(repository, tenant_id) diff = repository.get_shadow_diff(tenant_id=tenant_id, ledger_id=ledger.id, diff_key=diff_key) if diff is None: raise KnowledgeFSCutoverNotFoundError("Shadow authorization diff was not found") if diff.dify_allowed or diff.decision is KnowledgeFSShadowAuthorizationDecision.EXPANDED: raise KnowledgeFSCutoverGateBlockedError("Access-expanding diffs cannot be approved as fail-closed") changed = repository.set_shadow_diff_status( tenant_id=tenant_id, ledger_id=ledger.id, diff_key=diff_key, expected_status=KnowledgeFSMigrationIssueStatus.OPEN, new_status=KnowledgeFSMigrationIssueStatus.APPROVED_FAIL_CLOSED, account_id=account_id, changed_at=_naive_utc(approved_at), ) if not changed and diff.status is not KnowledgeFSMigrationIssueStatus.APPROVED_FAIL_CLOSED: raise KnowledgeFSCutoverConflictError("Shadow authorization diff changed during approval") def approve_issue_fail_closed( self, *, tenant_id: str, issue_key: str, account_id: str, approved_at: datetime, ) -> None: with self._session_maker.begin() as session: repository = SQLAlchemyKnowledgeFSCutoverRepository(session) ledger = self._require_ledger(repository, tenant_id) issue = repository.get_issue(tenant_id=tenant_id, ledger_id=ledger.id, issue_key=issue_key) if issue is None: raise KnowledgeFSCutoverNotFoundError("Migration issue was not found") approvable = issue.kind in { KnowledgeFSMigrationIssueKind.UNKNOWN_EXTERNAL_ACCESS, KnowledgeFSMigrationIssueKind.UNRESOLVED_SUBJECT, } and not issue.issue_key.startswith("owner:") if not approvable: raise KnowledgeFSCutoverGateBlockedError( "Only unknown fail-closed authorization can be approved; this issue must be resolved" ) changed = repository.set_issue_status( tenant_id=tenant_id, ledger_id=ledger.id, issue_key=issue_key, expected_status=KnowledgeFSMigrationIssueStatus.OPEN, new_status=KnowledgeFSMigrationIssueStatus.APPROVED_FAIL_CLOSED, account_id=account_id, changed_at=_naive_utc(approved_at), ) if not changed and issue.status is not KnowledgeFSMigrationIssueStatus.APPROVED_FAIL_CLOSED: raise KnowledgeFSCutoverConflictError("Migration issue changed during approval") def resolve_issue( self, *, tenant_id: str, issue_key: str, account_id: str, resolved_at: datetime, ) -> None: """Resolve remediated non-unknown evidence; unknown authorization requires fail-closed approval.""" with self._session_maker.begin() as session: repository = SQLAlchemyKnowledgeFSCutoverRepository(session) ledger = self._require_ledger(repository, tenant_id) issue = repository.get_issue(tenant_id=tenant_id, ledger_id=ledger.id, issue_key=issue_key) if issue is None: raise KnowledgeFSCutoverNotFoundError("Migration issue was not found") if issue.kind in { KnowledgeFSMigrationIssueKind.UNKNOWN_EXTERNAL_ACCESS, KnowledgeFSMigrationIssueKind.UNRESOLVED_SUBJECT, } and not issue.issue_key.startswith("owner:"): raise KnowledgeFSCutoverGateBlockedError( "Unknown authorization evidence can only be approved as fail-closed" ) if issue.status is KnowledgeFSMigrationIssueStatus.RESOLVED: return changed = repository.set_issue_status( tenant_id=tenant_id, ledger_id=ledger.id, issue_key=issue_key, expected_status=issue.status, new_status=KnowledgeFSMigrationIssueStatus.RESOLVED, account_id=account_id, changed_at=_naive_utc(resolved_at), ) if not changed: raise KnowledgeFSCutoverConflictError("Migration issue changed during resolution") def resolve_shadow_diff( self, *, tenant_id: str, diff_key: str, account_id: str, resolved_at: datetime, ) -> None: """Resolve a remediated known diff; unknown evidence remains approval-only.""" with self._session_maker.begin() as session: repository = SQLAlchemyKnowledgeFSCutoverRepository(session) ledger = self._require_ledger(repository, tenant_id) diff = repository.get_shadow_diff(tenant_id=tenant_id, ledger_id=ledger.id, diff_key=diff_key) if diff is None: raise KnowledgeFSCutoverNotFoundError("Shadow authorization diff was not found") if diff.decision is KnowledgeFSShadowAuthorizationDecision.UNKNOWN: raise KnowledgeFSCutoverGateBlockedError("Unknown shadow evidence can only be approved as fail-closed") if diff.decision is KnowledgeFSShadowAuthorizationDecision.EXPANDED: raise KnowledgeFSCutoverGateBlockedError( "Expanded shadow evidence requires a safe shadow re-evaluation" ) if diff.status is KnowledgeFSMigrationIssueStatus.RESOLVED: return changed = repository.set_shadow_diff_status( tenant_id=tenant_id, ledger_id=ledger.id, diff_key=diff_key, expected_status=diff.status, new_status=KnowledgeFSMigrationIssueStatus.RESOLVED, account_id=account_id, changed_at=_naive_utc(resolved_at), ) if not changed: raise KnowledgeFSCutoverConflictError("Shadow authorization diff changed during resolution") def resolve_quarantine( self, payload: QuarantineResolutionInput, *, apply: bool, ) -> QuarantineResolutionReport: """Resolve one quarantined source with immutable evidence and row-version CAS.""" tenant_id = str(payload.tenant_id) with self._session_maker.begin() as session: repository = SQLAlchemyKnowledgeFSCutoverRepository(session) ledger = self._require_ledger(repository, tenant_id) item = repository.get_quarantine( tenant_id=tenant_id, ledger_id=ledger.id, source_kind=payload.source_kind, source_id=payload.source_id, ) if item is None: raise KnowledgeFSCutoverNotFoundError("Quarantine item was not found") resolved_at = _naive_utc(payload.resolved_at) if resolved_at > _naive_utc(self._clock()): raise KnowledgeFSCutoverConflictError("Quarantine resolution timestamp cannot be in the future") evidence = cast(dict[str, object], payload.evidence.model_dump(mode="json")) if item.disposition is KnowledgeFSMigrationQuarantineDisposition.RESOLVED: if ( item.row_version != payload.expected_row_version + 1 or item.resolved_by_operator != payload.resolved_by_operator or item.resolved_by_account_id != str(payload.resolved_by_account_id) or item.evidence != evidence or item.resolved_at != resolved_at ): raise KnowledgeFSCutoverConflictError( "Quarantine item was already resolved with different evidence" ) return QuarantineResolutionReport( tenant_id, item.source_kind.value, item.source_id, item.disposition.value, item.row_version, apply, True, ) if ledger.phase in { KnowledgeFSWorkspaceCutoverPhase.CUTOVER, KnowledgeFSWorkspaceCutoverPhase.OBSERVING, KnowledgeFSWorkspaceCutoverPhase.READY_FOR_CLEANUP, }: raise KnowledgeFSCutoverConflictError("Quarantine resolution is closed after cutover") if item.row_version != payload.expected_row_version: raise KnowledgeFSCutoverConflictError("Quarantine item version changed") self._validate_quarantine_resolution(session, ledger, item, payload, resolved_at) if not apply: return QuarantineResolutionReport( tenant_id, item.source_kind.value, item.source_id, item.disposition.value, item.row_version, False, False, ) changed = repository.resolve_quarantine( KnowledgeFSQuarantineCASUpdate( tenant_id=tenant_id, ledger_id=ledger.id, source_kind=item.source_kind, source_id=item.source_id, expected_disposition=item.disposition, expected_row_version=item.row_version, resolved_by_operator=payload.resolved_by_operator, resolved_by_account_id=str(payload.resolved_by_account_id), evidence=evidence, resolved_at=resolved_at, ) ) if not changed: raise KnowledgeFSCutoverConflictError("Quarantine item changed during resolution") return QuarantineResolutionReport( tenant_id, item.source_kind.value, item.source_id, KnowledgeFSMigrationQuarantineDisposition.RESOLVED.value, item.row_version + 1, True, False, ) def _validate_quarantine_resolution( self, session: Session, ledger: KnowledgeFSWorkspaceCutoverLedger, item: KnowledgeFSMigrationQuarantine, payload: QuarantineResolutionInput, resolved_at: datetime, ) -> None: if isinstance(payload.evidence, LegacyCredentialRotationEvidence): credential_evidence = payload.evidence if credential_evidence.legacy_key_id != item.source_id: raise KnowledgeFSCutoverConflictError("Credential rotation evidence names a different legacy key") knowledge_space_id = str(credential_evidence.knowledge_space_id) if item.details.get("knowledge_space_id") != knowledge_space_id: raise KnowledgeFSCutoverConflictError( "Credential rotation evidence names a different KnowledgeFS Space" ) if _naive_utc(credential_evidence.legacy_revoked_at) > resolved_at: raise KnowledgeFSCutoverConflictError("Legacy credential revocation occurred after resolution") control_space = session.scalar( sa.select(KnowledgeFSControlSpace).where( KnowledgeFSControlSpace.id == str(credential_evidence.control_space_id), KnowledgeFSControlSpace.tenant_id == ledger.tenant_id, KnowledgeFSControlSpace.knowledge_space_id == knowledge_space_id, KnowledgeFSControlSpace.state == KnowledgeFSControlSpaceState.ACTIVE, ) ) if control_space is None: raise KnowledgeFSCutoverGateBlockedError( "Credential rotation requires the tenant-owned active control-space" ) credential = session.scalar( sa.select(KnowledgeFSApiCredential).where( KnowledgeFSApiCredential.id == str(credential_evidence.dify_credential_id), KnowledgeFSApiCredential.tenant_id == ledger.tenant_id, KnowledgeFSApiCredential.control_space_id == control_space.id, ) ) if ( credential is None or credential.status is not KnowledgeFSApiCredentialStatus.ACTIVE or credential.revision != credential_evidence.dify_credential_revision or (credential.expires_at is not None and credential.expires_at <= resolved_at) ): raise KnowledgeFSCutoverGateBlockedError( "Credential rotation requires a matching active Dify-managed credential" ) if credential.credential_prefix == item.details.get( "prefix" ) and credential.credential_last4 == item.details.get("last4"): raise KnowledgeFSCutoverGateBlockedError("Credential rotation did not replace the legacy credential") return if isinstance(payload.evidence, LegacyTaskResolutionEvidence): task_evidence = payload.evidence expected_actions = { KnowledgeFSMigrationQuarantineDisposition.MIGRATABLE: "migrate", KnowledgeFSMigrationQuarantineDisposition.WAIT_FOR_COMPLETION: "wait", KnowledgeFSMigrationQuarantineDisposition.CANCEL: "cancel", KnowledgeFSMigrationQuarantineDisposition.ISOLATE: "isolate", } expected_results = { "migrate": {"migrated"}, "wait": {"completed", "failed", "canceled"}, "cancel": {"canceled"}, "isolate": {"isolated"}, } if task_evidence.task_id != item.source_id or task_evidence.action != expected_actions.get( item.disposition ): raise KnowledgeFSCutoverConflictError("Task evidence does not match its classified disposition") if task_evidence.resulting_state not in expected_results[task_evidence.action]: raise KnowledgeFSCutoverConflictError("Task evidence has an invalid resulting state") if task_evidence.final_task_watermark < ledger.source_task_watermark: raise KnowledgeFSCutoverGateBlockedError("Task evidence is below the inventoried task watermark") def legacy_dependency_dashboard( self, *, tenant_id: str, dependencies: Iterable[LegacyDependencyInput], expected_cas_version: int | None, checked_at: datetime, apply: bool, ) -> LegacyDependencyDashboard: items = tuple(dependencies) active_rows = sum(item.active_rows for item in items) ready = active_rows == 0 report = LegacyDependencyDashboard( tenant_id, ready, sum(item.kind == "permission_snapshot" for item in items), sum(item.kind == "foreign_key" for item in items), active_rows, tuple(item.model_dump(mode="json") for item in items), apply, ) if not apply: return report if expected_cas_version is None: raise KnowledgeFSCutoverConflictError("Applying a dependency dashboard requires expected_cas_version") with self._session_maker.begin() as session: repository = SQLAlchemyKnowledgeFSCutoverRepository(session) ledger = self._require_ledger(repository, tenant_id) if ledger.cas_version != expected_cas_version: raise KnowledgeFSCutoverConflictError("Cutover ledger version changed") for item in items: if item.active_rows > 0: kind = ( KnowledgeFSMigrationIssueKind.LEGACY_SNAPSHOT_DEPENDENCY if item.kind == "permission_snapshot" else KnowledgeFSMigrationIssueKind.LEGACY_FOREIGN_KEY_DEPENDENCY ) self._ensure_issue( repository, ledger, issue_key=f"legacy:{item.dependency_key}", kind=kind, resource_type="legacy_dependency", resource_id=item.dependency_key, details=item.model_dump(mode="json"), ) dependency_keys = {f"legacy:{item.dependency_key}" for item in items if item.active_rows > 0} for issue in repository.list_issues(tenant_id=tenant_id, ledger_id=ledger.id): if ( issue.kind in { KnowledgeFSMigrationIssueKind.LEGACY_SNAPSHOT_DEPENDENCY, KnowledgeFSMigrationIssueKind.LEGACY_FOREIGN_KEY_DEPENDENCY, } and issue.issue_key not in dependency_keys and issue.status is not KnowledgeFSMigrationIssueStatus.RESOLVED ): repository.set_issue_status( tenant_id=tenant_id, ledger_id=ledger.id, issue_key=issue.issue_key, expected_status=issue.status, new_status=KnowledgeFSMigrationIssueStatus.RESOLVED, account_id=tenant_id, changed_at=_naive_utc(checked_at), ) self._cas( repository, KnowledgeFSCutoverCASUpdate( tenant_id=tenant_id, expected_phase=ledger.phase, expected_cas_version=ledger.cas_version, new_phase=ledger.phase, legacy_dependency_report=[dict(dependency) for dependency in report.dependencies], legacy_dependency_checked_at=_naive_utc(checked_at), legacy_dependency_ready=ready, ), ) return report def freeze( self, *, tenant_id: str, expected_cas_version: int, freeze_at: datetime ) -> KnowledgeFSWorkspaceCutoverLedger: freeze_time = _naive_utc(freeze_at) if freeze_time > _naive_utc(self._clock()): raise KnowledgeFSCutoverConflictError("Freeze timestamp cannot be in the future") # Snapshot all gates and the control identity, then release the transaction before the # remote call. Local FROZEN state is not visible until KnowledgeFS durably acknowledges. with self._session_maker.begin() as session: repository = SQLAlchemyKnowledgeFSCutoverRepository(session) ledger = self._require_ledger(repository, tenant_id) self._assert_phase(ledger, KnowledgeFSWorkspaceCutoverPhase.SHADOW, expected_cas_version) self._assert_shadow_evidence_complete(ledger) self._assert_no_open_gates(repository, ledger) self._assert_no_unresolved_quarantine(repository, ledger) if not ledger.legacy_dependency_ready or ledger.legacy_dependency_checked_at is None: raise KnowledgeFSCutoverGateBlockedError("Legacy snapshot/FK dependency dashboard is not ready") control_space = self._require_activation_control_space(session, tenant_id=tenant_id) freeze_request = _freeze_request(ledger, control_space_id=control_space.id) if self._remote_factory is None: raise KnowledgeFSCutoverGateBlockedError("KnowledgeFS remote freeze is not configured") try: remote = self._remote_factory() except RuntimeError as exc: raise KnowledgeFSCutoverGateBlockedError("KnowledgeFS remote freeze is not configured") from exc if remote is None: raise KnowledgeFSCutoverGateBlockedError("KnowledgeFS remote freeze is not configured") try: raw_ack = remote.freeze_dify_workspace_integration(freeze_request) except KnowledgeFSLifecycleRemoteError as exc: raise KnowledgeFSCutoverGateBlockedError("KnowledgeFS remote freeze was not acknowledged") from exc try: ack = ( raw_ack if isinstance(raw_ack, KnowledgeFSDifyIntegrationFreezeAck) else KnowledgeFSDifyIntegrationFreezeAck.model_validate(raw_ack, from_attributes=True) ) except (TypeError, ValueError, ValidationError) as exc: raise KnowledgeFSCutoverGateBlockedError("KnowledgeFS remote freeze acknowledgement was malformed") from exc _assert_exact_freeze_ack(freeze_request, ack) acknowledged_at = _naive_utc(self._clock()) with self._session_maker.begin() as session: repository = SQLAlchemyKnowledgeFSCutoverRepository(session) ledger = self._require_ledger(repository, tenant_id) self._assert_phase(ledger, KnowledgeFSWorkspaceCutoverPhase.SHADOW, expected_cas_version) self._assert_shadow_evidence_complete(ledger) self._assert_no_open_gates(repository, ledger) self._assert_no_unresolved_quarantine(repository, ledger) if not ledger.legacy_dependency_ready or ledger.legacy_dependency_checked_at is None: raise KnowledgeFSCutoverGateBlockedError("Legacy snapshot/FK dependency dashboard is not ready") control_space = self._require_activation_control_space( session, tenant_id=tenant_id, control_space_id=freeze_request.control_space_id, ) if _freeze_request(ledger, control_space_id=control_space.id) != freeze_request: raise KnowledgeFSCutoverConflictError("Freeze evidence changed after remote acknowledgement") self._cas( repository, KnowledgeFSCutoverCASUpdate( tenant_id=tenant_id, expected_phase=ledger.phase, expected_cas_version=ledger.cas_version, new_phase=KnowledgeFSWorkspaceCutoverPhase.FROZEN, freeze_at=freeze_time, remote_freeze_id=ack.freeze_id, remote_freeze_revision=ack.freeze_revision, remote_freeze_digest=ack.source_revision_digest, remote_freeze_task_watermark=ack.source_task_watermark, remote_freeze_control_space_id=control_space.id, remote_freeze_frozen_at=_naive_utc(ack.frozen_at), remote_freeze_updated_at=_naive_utc(ack.updated_at), remote_freeze_acknowledged_at=acknowledged_at, remote_freeze_applied=ack.applied, remote_freeze_replayed=ack.replayed, ), ) return self._require_ledger(repository, tenant_id) def apply_final_delta(self, payload: FinalDeltaInput) -> KnowledgeFSWorkspaceCutoverLedger: tenant_id = str(payload.tenant_id) with self._session_maker.begin() as session: repository = SQLAlchemyKnowledgeFSCutoverRepository(session) ledger = self._require_ledger(repository, tenant_id) self._assert_phase(ledger, KnowledgeFSWorkspaceCutoverPhase.FROZEN, payload.expected_cas_version) final_watermark = payload.final_revision_watermark.to_record() applied_watermark = payload.applied_revision_watermark.to_record() if not _watermark_at_least(final_watermark, ledger.source_revision_watermark): raise KnowledgeFSCutoverConflictError("Final authorization watermark moved backwards") if payload.final_task_watermark < ledger.source_task_watermark: raise KnowledgeFSCutoverConflictError("Final task watermark moved backwards") self._cas( repository, KnowledgeFSCutoverCASUpdate( tenant_id=tenant_id, expected_phase=ledger.phase, expected_cas_version=ledger.cas_version, new_phase=ledger.phase, final_revision_watermark=final_watermark, applied_revision_watermark=applied_watermark, final_task_watermark=payload.final_task_watermark, applied_task_watermark=payload.applied_task_watermark, ), ) return self._require_ledger(repository, tenant_id) def cutover( self, *, tenant_id: str, expected_cas_version: int, cutover_at: datetime, rollback_cutoff_at: datetime, ) -> KnowledgeFSWorkspaceCutoverLedger: cutover_time = _naive_utc(cutover_at) cutoff = _naive_utc(rollback_cutoff_at) if cutoff <= cutover_time: raise KnowledgeFSCutoverConflictError("Rollback cutoff must be after cutover") # The first short transaction snapshots all gate and activation evidence. It # deliberately commits before capability issuance or any remote HTTP call. with self._session_maker.begin() as session: repository = SQLAlchemyKnowledgeFSCutoverRepository(session) ledger = self._require_ledger(repository, tenant_id) self._assert_cutover_ready(repository, ledger, expected_cas_version) control_space = self._require_activation_control_space(session, tenant_id=tenant_id) activation_request = _activation_request(ledger, control_space_id=control_space.id) if self._remote_factory is None: raise KnowledgeFSCutoverGateBlockedError("KnowledgeFS remote activation is not configured") try: remote = self._remote_factory() except RuntimeError as exc: raise KnowledgeFSCutoverGateBlockedError("KnowledgeFS remote activation is not configured") from exc if remote is None: raise KnowledgeFSCutoverGateBlockedError("KnowledgeFS remote activation is not configured") try: raw_ack = remote.activate_dify_workspace_integration(activation_request) except KnowledgeFSLifecycleRemoteError as exc: raise KnowledgeFSCutoverGateBlockedError("KnowledgeFS remote activation was not acknowledged") from exc try: ack = ( raw_ack if isinstance(raw_ack, KnowledgeFSDifyIntegrationActivationAck) else KnowledgeFSDifyIntegrationActivationAck.model_validate(raw_ack, from_attributes=True) ) except (TypeError, ValueError, ValidationError) as exc: raise KnowledgeFSCutoverGateBlockedError( "KnowledgeFS remote activation acknowledgement was malformed" ) from exc _assert_exact_activation_ack(activation_request, ack) acknowledged_at = _naive_utc(self._clock()) # Re-open a transaction only after the exact durable ACK. Re-evaluate every # gate and identity, then publish all switches and ACK evidence in one CAS. with self._session_maker.begin() as session: repository = SQLAlchemyKnowledgeFSCutoverRepository(session) ledger = self._require_ledger(repository, tenant_id) self._assert_cutover_ready(repository, ledger, expected_cas_version) control_space = self._require_activation_control_space( session, tenant_id=tenant_id, control_space_id=activation_request.control_space_id, ) if _activation_request(ledger, control_space_id=control_space.id) != activation_request: raise KnowledgeFSCutoverConflictError( "Cutover activation evidence changed after remote acknowledgement" ) self._cas( repository, KnowledgeFSCutoverCASUpdate( tenant_id=tenant_id, expected_phase=ledger.phase, expected_cas_version=ledger.cas_version, new_phase=KnowledgeFSWorkspaceCutoverPhase.CUTOVER, cutover_at=cutover_time, rollback_cutoff_at=cutoff, product_routes_enabled=True, capability_v2_enabled=True, integrated_mode_enabled=True, legacy_acl_read_only=True, clear_smoke_results=True, remote_activation_id=ack.activation_id, remote_activation_revision=ack.activation_revision, remote_activation_digest=ack.source_revision_digest, remote_activation_control_space_id=control_space.id, remote_activation_activated_at=_naive_utc(ack.activated_at), remote_activation_updated_at=_naive_utc(ack.updated_at), remote_activation_acknowledged_at=acknowledged_at, remote_activation_applied=ack.applied, remote_activation_replayed=ack.replayed, ), ) return self._require_ledger(repository, tenant_id) def record_smoke_results( self, *, tenant_id: str, expected_cas_version: int, results: CutoverSmokeResultsInput, ) -> KnowledgeFSWorkspaceCutoverLedger: if str(results.tenant_id) != tenant_id: raise KnowledgeFSCutoverConflictError("Smoke evidence belongs to a different Workspace") observed_at = _naive_utc(results.observed_at) if observed_at > _naive_utc(self._clock()): raise KnowledgeFSCutoverConflictError("Smoke evidence timestamp cannot be in the future") with self._session_maker.begin() as session: repository = SQLAlchemyKnowledgeFSCutoverRepository(session) ledger = self._require_ledger(repository, tenant_id) self._assert_phase(ledger, KnowledgeFSWorkspaceCutoverPhase.CUTOVER, expected_cas_version) if ledger.cutover_at is None or observed_at < ledger.cutover_at: raise KnowledgeFSCutoverConflictError("Smoke evidence predates Workspace cutover") self._cas( repository, KnowledgeFSCutoverCASUpdate( tenant_id=tenant_id, expected_phase=ledger.phase, expected_cas_version=ledger.cas_version, new_phase=ledger.phase, smoke_results=results.to_record(), ), ) return self._require_ledger(repository, tenant_id) def begin_observation( self, *, tenant_id: str, expected_cas_version: int, started_at: datetime, window_ends_at: datetime, maximum_task_expires_at: datetime, ) -> KnowledgeFSWorkspaceCutoverLedger: started = _naive_utc(started_at) window_end = _naive_utc(window_ends_at) task_expiry = _naive_utc(maximum_task_expires_at) if window_end <= started: raise KnowledgeFSCutoverConflictError("Observation window must end after it starts") with self._session_maker.begin() as session: repository = SQLAlchemyKnowledgeFSCutoverRepository(session) ledger = self._require_ledger(repository, tenant_id) self._assert_phase(ledger, KnowledgeFSWorkspaceCutoverPhase.CUTOVER, expected_cas_version) if not knowledge_fs_cutover_smoke_results_passed(ledger.smoke_results): raise KnowledgeFSCutoverGateBlockedError("All cutover smoke checks must pass before observation") self._cas( repository, KnowledgeFSCutoverCASUpdate( tenant_id=tenant_id, expected_phase=ledger.phase, expected_cas_version=ledger.cas_version, new_phase=KnowledgeFSWorkspaceCutoverPhase.OBSERVING, observation_started_at=started, observation_window_ends_at=window_end, maximum_task_expires_at=task_expiry, ), ) return self._require_ledger(repository, tenant_id) def complete_observation( self, *, tenant_id: str, expected_cas_version: int, observed_at: datetime ) -> KnowledgeFSWorkspaceCutoverLedger: observed = _naive_utc(observed_at) with self._session_maker.begin() as session: repository = SQLAlchemyKnowledgeFSCutoverRepository(session) ledger = self._require_ledger(repository, tenant_id) self._assert_phase(ledger, KnowledgeFSWorkspaceCutoverPhase.OBSERVING, expected_cas_version) if ( ledger.observation_window_ends_at is None or observed < ledger.observation_window_ends_at or ledger.maximum_task_expires_at is None or observed < ledger.maximum_task_expires_at ): raise KnowledgeFSCutoverGateBlockedError( "Observation window and maximum task TTL must both elapse before cleanup" ) self._cas( repository, KnowledgeFSCutoverCASUpdate( tenant_id=tenant_id, expected_phase=ledger.phase, expected_cas_version=ledger.cas_version, new_phase=KnowledgeFSWorkspaceCutoverPhase.READY_FOR_CLEANUP, observation_completed_at=observed, ), ) return self._require_ledger(repository, tenant_id) def rollback( self, *, tenant_id: str, expected_cas_version: int, rolled_back_at: datetime ) -> KnowledgeFSWorkspaceCutoverLedger: rollback_time = _naive_utc(rolled_back_at) now = _naive_utc(self._clock()) with self._session_maker.begin() as session: repository = SQLAlchemyKnowledgeFSCutoverRepository(session) ledger = self._require_ledger(repository, tenant_id) if ledger.phase not in { KnowledgeFSWorkspaceCutoverPhase.CUTOVER, KnowledgeFSWorkspaceCutoverPhase.OBSERVING, KnowledgeFSWorkspaceCutoverPhase.READY_FOR_CLEANUP, }: raise KnowledgeFSCutoverConflictError("Rollback is only available after cutover and before cleanup") if ledger.cas_version != expected_cas_version: raise KnowledgeFSCutoverConflictError("Cutover ledger version changed") if ( ledger.rollback_cutoff_at is None or rollback_time >= ledger.rollback_cutoff_at or now >= ledger.rollback_cutoff_at ): raise KnowledgeFSCutoverGateBlockedError("Rollback cutoff has elapsed") if rollback_time > now: raise KnowledgeFSCutoverConflictError("Rollback timestamp cannot be in the future") if ledger.irreversible_cleanup_at is not None: raise KnowledgeFSCutoverGateBlockedError("Irreversible cleanup has already started") # Legacy ACL remains read-only: post-cutover rollback is a safe traffic stop, # never a return to the stale KnowledgeFS authorization source. Remote # activation and its acknowledgement are intentionally immutable here. self._cas( repository, KnowledgeFSCutoverCASUpdate( tenant_id=tenant_id, expected_phase=ledger.phase, expected_cas_version=ledger.cas_version, new_phase=KnowledgeFSWorkspaceCutoverPhase.FROZEN, rolled_back_at=rollback_time, product_routes_enabled=False, ), ) return self._require_ledger(repository, tenant_id) def status(self, *, tenant_id: str) -> dict[str, object]: with self._session_maker() as session: repository = SQLAlchemyKnowledgeFSCutoverRepository(session) ledger = self._require_ledger(repository, tenant_id) issues = repository.list_issues(tenant_id=tenant_id, ledger_id=ledger.id) quarantine = repository.list_quarantine(tenant_id=tenant_id, ledger_id=ledger.id) shadow_diffs = repository.list_shadow_diffs(tenant_id=tenant_id, ledger_id=ledger.id) shadow_observations = repository.list_shadow_observations(tenant_id=tenant_id, ledger_id=ledger.id) return { "tenant_id": ledger.tenant_id, "phase": ledger.phase.value, "cas_version": ledger.cas_version, "source_revision_watermark": ledger.source_revision_watermark, "final_revision_watermark": ledger.final_revision_watermark, "applied_revision_watermark": ledger.applied_revision_watermark, "source_task_watermark": ledger.source_task_watermark, "final_task_watermark": ledger.final_task_watermark, "applied_task_watermark": ledger.applied_task_watermark, "feature_state": { "product_routes_enabled": ledger.product_routes_enabled, "capability_v2_enabled": ledger.capability_v2_enabled, "integrated_mode_enabled": ledger.integrated_mode_enabled, "legacy_acl_read_only": ledger.legacy_acl_read_only, }, "smoke_results": ledger.smoke_results, "legacy_dependency_report": ledger.legacy_dependency_report, "legacy_dependency_ready": ledger.legacy_dependency_ready, "shadow_started_at": _iso(ledger.shadow_started_at), "shadow_completed_at": _iso(ledger.shadow_completed_at), "shadow_evidence_digest": ledger.shadow_evidence_digest, "shadow_observation_count": ledger.shadow_observation_count, "shadow_window_started_at": _iso(ledger.shadow_window_started_at), "shadow_window_ended_at": _iso(ledger.shadow_window_ended_at), "shadow_traffic_zero": ledger.shadow_traffic_zero, "shadow_traffic_zero_evidence": ledger.shadow_traffic_zero_evidence, "shadow_latest_observed_revision": ledger.shadow_latest_observed_revision, "shadow_producer": ledger.shadow_producer, "shadow_completed_by_operator": ledger.shadow_completed_by_operator, "shadow_completed_by_account_id": ledger.shadow_completed_by_account_id, "remote_freeze_id": ledger.remote_freeze_id, "remote_freeze_revision": ledger.remote_freeze_revision, "remote_freeze_digest": ledger.remote_freeze_digest, "remote_freeze_task_watermark": ledger.remote_freeze_task_watermark, "remote_freeze_control_space_id": ledger.remote_freeze_control_space_id, "remote_freeze_frozen_at": _iso(ledger.remote_freeze_frozen_at), "remote_freeze_updated_at": _iso(ledger.remote_freeze_updated_at), "remote_freeze_acknowledged_at": _iso(ledger.remote_freeze_acknowledged_at), "remote_freeze_applied": ledger.remote_freeze_applied, "remote_freeze_replayed": ledger.remote_freeze_replayed, "remote_activation_id": ledger.remote_activation_id, "remote_activation_revision": ledger.remote_activation_revision, "remote_activation_digest": ledger.remote_activation_digest, "remote_activation_control_space_id": ledger.remote_activation_control_space_id, "remote_activation_activated_at": _iso(ledger.remote_activation_activated_at), "remote_activation_updated_at": _iso(ledger.remote_activation_updated_at), "remote_activation_acknowledged_at": _iso(ledger.remote_activation_acknowledged_at), "remote_activation_applied": ledger.remote_activation_applied, "remote_activation_replayed": ledger.remote_activation_replayed, "open_issues": sum(issue.status is KnowledgeFSMigrationIssueStatus.OPEN for issue in issues), "open_shadow_diffs": sum(diff.status is KnowledgeFSMigrationIssueStatus.OPEN for diff in shadow_diffs), "quarantine_count": len(quarantine), "unresolved_cutover_quarantine": sum( item.source_kind in _CUTOVER_QUARANTINE_KINDS and item.disposition is not KnowledgeFSMigrationQuarantineDisposition.RESOLVED for item in quarantine ), "issues": [ { "issue_key": issue.issue_key, "kind": issue.kind.value, "status": issue.status.value, "resource_type": issue.resource_type, "resource_id": issue.resource_id, "details": issue.details, "approved_by_account_id": issue.approved_by_account_id, "approved_at": _iso(issue.approved_at), "resolved_by_account_id": issue.resolved_by_account_id, "resolved_at": _iso(issue.resolved_at), } for issue in issues ], "quarantine": [ { "source_kind": item.source_kind.value, "source_id": item.source_id, "reason_code": item.reason_code, "disposition": item.disposition.value, "details": item.details, "resolved_by_operator": item.resolved_by_operator, "resolved_by_account_id": item.resolved_by_account_id, "evidence": item.evidence, "resolved_at": _iso(item.resolved_at), "row_version": item.row_version, } for item in quarantine ], "shadow_diffs": [ { "diff_key": diff.diff_key, "control_space_id": diff.control_space_id, "principal": diff.principal, "legacy_allowed": diff.legacy_allowed, "dify_allowed": diff.dify_allowed, "decision": diff.decision.value, "reason": diff.reason, "observed_revision": diff.observed_revision, "status": diff.status.value, "approved_by_account_id": diff.approved_by_account_id, "approved_at": _iso(diff.approved_at), "resolved_by_account_id": diff.resolved_by_account_id, "resolved_at": _iso(diff.resolved_at), "current_evidence_digest": diff.current_evidence_digest, "last_observed_at": _iso(diff.last_observed_at), "row_version": diff.row_version, } for diff in shadow_diffs ], "shadow_observations": [ { "diff_key": item.diff_key, "producer": item.producer, "control_space_id": item.control_space_id, "principal": item.principal, "legacy_allowed": item.legacy_allowed, "dify_allowed": item.dify_allowed, "decision": item.decision.value, "reason": item.reason, "observed_revision": item.observed_revision, "observed_at": _iso(item.observed_at), "evidence_digest": item.evidence_digest, } for item in shadow_observations ], "freeze_at": _iso(ledger.freeze_at), "cutover_at": _iso(ledger.cutover_at), "rolled_back_at": _iso(ledger.rolled_back_at), "rollback_cutoff_at": _iso(ledger.rollback_cutoff_at), "observation_started_at": _iso(ledger.observation_started_at), "observation_window_ends_at": _iso(ledger.observation_window_ends_at), "maximum_task_expires_at": _iso(ledger.maximum_task_expires_at), "observation_completed_at": _iso(ledger.observation_completed_at), "legacy_dependency_checked_at": _iso(ledger.legacy_dependency_checked_at), "irreversible_cleanup_at": _iso(ledger.irreversible_cleanup_at), } @staticmethod def _task_disposition_counts(spaces: Iterable[LegacySpaceInventoryInput]) -> dict[str, int]: counts = {disposition.value: 0 for disposition in KnowledgeFSMigrationQuarantineDisposition} for space in spaces: for task in space.tasks: counts[_classify_task(task).value] += 1 return {key: value for key, value in counts.items() if value > 0} @staticmethod def _assert_unique_inventory(spaces: Iterable[LegacySpaceInventoryInput]) -> None: knowledge_space_ids: set[UUID] = set() provisioning_keys: set[str] = set() for space in spaces: if space.knowledge_space_id in knowledge_space_ids: raise KnowledgeFSCutoverConflictError("Inventory contains a duplicate KnowledgeFS Space ID") if space.provisioning_key in provisioning_keys: raise KnowledgeFSCutoverConflictError("Inventory contains a duplicate provisioning key") knowledge_space_ids.add(space.knowledge_space_id) provisioning_keys.add(space.provisioning_key) resolved_permissions: set[UUID] = set() for permission in space.permissions: if permission.account_id is None: continue if permission.account_id in resolved_permissions: raise KnowledgeFSCutoverConflictError("Inventory contains duplicate resolved permission subjects") if permission.account_id == space.owner_account_id and permission.role != "owner": raise KnowledgeFSCutoverConflictError("Inventory cannot downgrade the registered owner") resolved_permissions.add(permission.account_id) @staticmethod def _project_backfill(payload: WorkspaceInventoryInput) -> CutoverBackfillReport: registered = sum(space.owner_account_id is not None for space in payload.spaces) granted = sum( (1 if space.owner_account_id is not None else 0) + sum(permission.account_id is not None for permission in space.permissions) for space in payload.spaces ) quarantined = sum( (space.owner_account_id is None) + sum(permission.account_id is None for permission in space.permissions) + len(space.legacy_api_keys) + sum( _classify_task(task) is not KnowledgeFSMigrationQuarantineDisposition.MIGRATABLE for task in space.tasks ) + len(space.orphan_resource_ids) for space in payload.spaces ) open_issues = sum( (space.owner_account_id is None) + sum(permission.account_id is None for permission in space.permissions) + (space.external_access_enabled is None or space.visibility == "unknown") for space in payload.spaces ) return CutoverBackfillReport( str(payload.tenant_id), registered, granted, quarantined, open_issues, KnowledgeFSWorkspaceCutoverPhase.INVENTORY.value, False, ) def _backfill_space( self, session: Session, repository: SQLAlchemyKnowledgeFSCutoverRepository, ledger: KnowledgeFSWorkspaceCutoverLedger, space: LegacySpaceInventoryInput, revision_watermark: KnowledgeFSCutoverRevisionWatermark, ) -> tuple[int, int, int]: knowledge_space_id = str(space.knowledge_space_id) quarantined = self._backfill_auxiliary_inventory(repository, ledger, space, knowledge_space_id) if space.owner_account_id is None: self._ensure_issue( repository, ledger, issue_key=f"owner:{knowledge_space_id}", kind=KnowledgeFSMigrationIssueKind.UNRESOLVED_SUBJECT, resource_type="knowledge_space", resource_id=knowledge_space_id, details={"subject_id": space.owner_subject_id}, ) self._ensure_quarantine( repository, ledger, source_kind=KnowledgeFSMigrationQuarantineKind.CONTROL_SPACE, source_id=knowledge_space_id, reason_code="UNRESOLVED_OWNER", disposition=KnowledgeFSMigrationQuarantineDisposition.PENDING, details={"subject_id": space.owner_subject_id}, ) return 0, 0, quarantined + 1 tenant_id = ledger.tenant_id owner_account_id = str(space.owner_account_id) existing = session.scalar( sa.select(KnowledgeFSControlSpace).where( KnowledgeFSControlSpace.tenant_id == tenant_id, (KnowledgeFSControlSpace.knowledge_space_id == knowledge_space_id) | (KnowledgeFSControlSpace.provisioning_key == space.provisioning_key), ) ) registered = 0 if existing is not None and ( existing.knowledge_space_id != knowledge_space_id or existing.provisioning_key != space.provisioning_key or existing.owner_account_id != owner_account_id ): self._ensure_issue( repository, ledger, issue_key=f"registration:{knowledge_space_id}", kind=KnowledgeFSMigrationIssueKind.REGISTRATION_CONFLICT, resource_type="knowledge_space", resource_id=knowledge_space_id, details={"provisioning_key": space.provisioning_key}, ) return 0, 0, quarantined visibility = _VISIBILITY_MAP.get(space.visibility, KnowledgeFSControlSpaceVisibility.ONLY_ME) if existing is None: existing = KnowledgeFSControlSpace( tenant_id=tenant_id, owner_account_id=owner_account_id, provisioning_key=space.provisioning_key, knowledge_space_id=knowledge_space_id, knowledge_space_revision=space.knowledge_space_revision, visibility=visibility, state=KnowledgeFSControlSpaceState.ACTIVE, lifecycle_operation_id="migration-backfill", ) session.add(existing) session.flush() registered = 1 elif existing.knowledge_space_revision > space.knowledge_space_revision: self._ensure_issue( repository, ledger, issue_key=f"revision-drift:{knowledge_space_id}", kind=KnowledgeFSMigrationIssueKind.REVISION_DRIFT, resource_type="knowledge_space", resource_id=knowledge_space_id, details={ "existing_revision": existing.knowledge_space_revision, "inventory_revision": space.knowledge_space_revision, }, ) return registered, 0, quarantined else: if ( existing.knowledge_space_revision != space.knowledge_space_revision or existing.visibility is not visibility ): existing.resource_version += 1 existing.knowledge_space_revision = space.knowledge_space_revision existing.visibility = visibility revision = session.scalar( sa.select(KnowledgeFSAuthorizationRevision).where( KnowledgeFSAuthorizationRevision.tenant_id == tenant_id, KnowledgeFSAuthorizationRevision.control_space_id == existing.id, ) ) if revision is None: revision = KnowledgeFSAuthorizationRevision( tenant_id=tenant_id, control_space_id=existing.id, ) session.add(revision) revision.membership_epoch = revision_watermark["membership_epoch"] revision.space_acl_epoch = revision_watermark["space_acl_epoch"] revision.external_access_epoch = revision_watermark["external_access_epoch"] revision.content_policy_revision = revision_watermark["content_policy_revision"] granted = self._ensure_permission( session, tenant_id=tenant_id, control_space=existing, account_id=owner_account_id, role=KnowledgeFSControlSpacePermissionRole.OWNER, ) for permission in space.permissions: if permission.account_id is None: continue granted += self._ensure_permission( session, tenant_id=tenant_id, control_space=existing, account_id=str(permission.account_id), role=KnowledgeFSControlSpacePermissionRole(permission.role), ) external_enabled = space.external_access_enabled is True external = session.scalar( sa.select(KnowledgeFSExternalAccessPolicy).where( KnowledgeFSExternalAccessPolicy.tenant_id == tenant_id, KnowledgeFSExternalAccessPolicy.control_space_id == existing.id, ) ) if external is None: external = KnowledgeFSExternalAccessPolicy( tenant_id=tenant_id, control_space_id=existing.id, ) session.add(external) elif any( ( external.service_api_enabled != external_enabled, external.agent_enabled != external_enabled, external.workflow_enabled != external_enabled, external.mcp_enabled, ) ): external.revision += 1 external.service_api_enabled = external_enabled external.agent_enabled = external_enabled external.workflow_enabled = external_enabled external.mcp_enabled = False return registered, granted, quarantined def _backfill_auxiliary_inventory( self, repository: SQLAlchemyKnowledgeFSCutoverRepository, ledger: KnowledgeFSWorkspaceCutoverLedger, space: LegacySpaceInventoryInput, knowledge_space_id: str, ) -> int: quarantined = 0 visibility = _VISIBILITY_MAP.get(space.visibility, KnowledgeFSControlSpaceVisibility.ONLY_ME) for permission in space.permissions: if permission.account_id is not None: continue self._ensure_issue( repository, ledger, issue_key=f"subject:{knowledge_space_id}:{permission.subject_id}", kind=KnowledgeFSMigrationIssueKind.UNRESOLVED_SUBJECT, resource_type="subject", resource_id=permission.subject_id, details={"knowledge_space_id": knowledge_space_id}, ) self._ensure_quarantine( repository, ledger, source_kind=KnowledgeFSMigrationQuarantineKind.SUBJECT, source_id=f"{knowledge_space_id}:{permission.subject_id}", reason_code="UNRESOLVED_SUBJECT", disposition=KnowledgeFSMigrationQuarantineDisposition.ISOLATE, details={"role": permission.role}, ) quarantined += 1 if space.external_access_enabled is None or space.visibility == "unknown": self._ensure_issue( repository, ledger, issue_key=f"unknown-access:{knowledge_space_id}", kind=KnowledgeFSMigrationIssueKind.UNKNOWN_EXTERNAL_ACCESS, resource_type="knowledge_space", resource_id=knowledge_space_id, details={"effective_external_access": False, "effective_visibility": visibility.value}, ) for legacy_key in space.legacy_api_keys: self._ensure_quarantine( repository, ledger, source_kind=KnowledgeFSMigrationQuarantineKind.LEGACY_API_KEY, source_id=legacy_key.key_id, reason_code="ROTATION_REQUIRED", disposition=KnowledgeFSMigrationQuarantineDisposition.ROTATE_CREDENTIAL, details={ "knowledge_space_id": knowledge_space_id, "prefix": legacy_key.prefix, "last4": legacy_key.last4, }, ) quarantined += 1 for task in space.tasks: disposition = _classify_task(task) self._ensure_quarantine( repository, ledger, source_kind=KnowledgeFSMigrationQuarantineKind.TASK, source_id=task.task_id, reason_code="LEGACY_TASK_CLASSIFICATION", disposition=disposition, details={ "knowledge_space_id": knowledge_space_id, "state": task.state, "subject_id": task.subject_id, "account_id": str(task.account_id) if task.account_id else None, "expires_at": task.expires_at.isoformat() if task.expires_at else None, }, ) quarantined += disposition is not KnowledgeFSMigrationQuarantineDisposition.MIGRATABLE for orphan_id in space.orphan_resource_ids: self._ensure_quarantine( repository, ledger, source_kind=KnowledgeFSMigrationQuarantineKind.ORPHAN_RESOURCE, source_id=orphan_id, reason_code="ORPHAN_RESOURCE", disposition=KnowledgeFSMigrationQuarantineDisposition.PENDING, details={"knowledge_space_id": knowledge_space_id}, ) quarantined += 1 return quarantined @staticmethod def _ensure_permission( session: Session, *, tenant_id: str, control_space: KnowledgeFSControlSpace, account_id: str, role: KnowledgeFSControlSpacePermissionRole, ) -> int: existing = session.scalar( sa.select(KnowledgeFSControlSpacePermission).where( KnowledgeFSControlSpacePermission.tenant_id == tenant_id, KnowledgeFSControlSpacePermission.control_space_id == control_space.id, KnowledgeFSControlSpacePermission.account_id == account_id, ) ) if existing is not None: if existing.role is not role: existing.role = role existing.revision += 1 return 0 session.add( KnowledgeFSControlSpacePermission( tenant_id=tenant_id, control_space_id=control_space.id, account_id=account_id, role=role, granted_by_account_id=control_space.owner_account_id, ) ) return 1 @staticmethod def _ensure_issue( repository: SQLAlchemyKnowledgeFSCutoverRepository, ledger: KnowledgeFSWorkspaceCutoverLedger, *, issue_key: str, kind: KnowledgeFSMigrationIssueKind, resource_type: str, resource_id: str, details: dict[str, object], ) -> None: if repository.get_issue(tenant_id=ledger.tenant_id, ledger_id=ledger.id, issue_key=issue_key) is None: repository.add_issue( KnowledgeFSMigrationIssue( tenant_id=ledger.tenant_id, ledger_id=ledger.id, issue_key=issue_key, kind=kind, resource_type=resource_type, resource_id=resource_id, details=details, ) ) @staticmethod def _ensure_quarantine( repository: SQLAlchemyKnowledgeFSCutoverRepository, ledger: KnowledgeFSWorkspaceCutoverLedger, *, source_kind: KnowledgeFSMigrationQuarantineKind, source_id: str, reason_code: str, disposition: KnowledgeFSMigrationQuarantineDisposition, details: dict[str, object], ) -> None: if ( repository.get_quarantine( tenant_id=ledger.tenant_id, ledger_id=ledger.id, source_kind=source_kind, source_id=source_id, ) is None ): repository.add_quarantine( KnowledgeFSMigrationQuarantine( tenant_id=ledger.tenant_id, ledger_id=ledger.id, source_kind=source_kind, source_id=source_id, reason_code=reason_code, disposition=disposition, details=details, ) ) def _validate_shadow_observation( self, ledger: KnowledgeFSWorkspaceCutoverLedger, item: ShadowAuthorizationObservationInput, ) -> None: if item.producer not in self._trusted_shadow_producers: raise KnowledgeFSCutoverGateBlockedError("Shadow observation producer is not trusted") if ledger.shadow_started_at is None: raise KnowledgeFSCutoverGateBlockedError("Shadow start evidence is missing") observed_at = _naive_utc(item.observed_at) if observed_at > _naive_utc(self._clock()): raise KnowledgeFSCutoverConflictError("Shadow observation timestamp cannot be in the future") if observed_at < ledger.shadow_started_at: raise KnowledgeFSCutoverConflictError("Shadow observation predates shadow start") required_revision = ledger.final_revision_watermark or ledger.source_revision_watermark if not _watermark_at_least(item.observed_revision.to_record(), required_revision): raise KnowledgeFSCutoverConflictError("Shadow observation revision is stale") @staticmethod def _record_shadow_observation( repository: SQLAlchemyKnowledgeFSCutoverRepository, ledger: KnowledgeFSWorkspaceCutoverLedger, item: ShadowAuthorizationObservationInput, decision: KnowledgeFSShadowAuthorizationDecision, *, apply: bool, ) -> bool: evidence_digest = _shadow_observation_digest(item) replay = repository.get_shadow_observation( tenant_id=ledger.tenant_id, ledger_id=ledger.id, diff_key=item.diff_key, evidence_digest=evidence_digest, ) if replay is not None: return True if ledger.shadow_completed_at is not None: raise KnowledgeFSCutoverConflictError("Shadow evidence is already complete") existing = repository.get_shadow_diff( tenant_id=ledger.tenant_id, ledger_id=ledger.id, diff_key=item.diff_key, ) dify_allowed = False if item.legacy_allowed is None else item.dify_allowed if existing is not None: safe_reevaluation = ( existing.decision in { KnowledgeFSShadowAuthorizationDecision.EXPANDED, KnowledgeFSShadowAuthorizationDecision.UNKNOWN, } and decision in { KnowledgeFSShadowAuthorizationDecision.MATCH, KnowledgeFSShadowAuthorizationDecision.TIGHTENED, } and _watermark_at_least(item.observed_revision.to_record(), existing.observed_revision) ) if not safe_reevaluation: raise KnowledgeFSCutoverConflictError("Shadow diff key was reused with different evidence") if not apply: return False repository.add_shadow_observation( _shadow_observation_model(ledger=ledger, item=item, decision=decision, digest=evidence_digest) ) changed = repository.reevaluate_shadow_diff( KnowledgeFSShadowDiffCASUpdate( tenant_id=ledger.tenant_id, ledger_id=ledger.id, diff_key=item.diff_key, expected_row_version=existing.row_version, control_space_id=str(item.control_space_id) if item.control_space_id else None, principal=item.principal, legacy_allowed=item.legacy_allowed, dify_allowed=dify_allowed, decision=decision, reason=item.reason, observed_revision=item.observed_revision.to_record(), status=KnowledgeFSMigrationIssueStatus.RESOLVED, current_evidence_digest=evidence_digest, last_observed_at=_naive_utc(item.observed_at), ) ) if not changed: raise KnowledgeFSCutoverConflictError("Shadow diff changed during re-evaluation") return False if not apply: return False repository.add_shadow_observation( _shadow_observation_model(ledger=ledger, item=item, decision=decision, digest=evidence_digest) ) repository.add_shadow_diff( KnowledgeFSShadowAuthorizationDiff( tenant_id=ledger.tenant_id, ledger_id=ledger.id, diff_key=item.diff_key, control_space_id=str(item.control_space_id) if item.control_space_id else None, principal=item.principal, legacy_allowed=item.legacy_allowed, dify_allowed=dify_allowed, decision=decision, reason=item.reason, observed_revision=item.observed_revision.to_record(), current_evidence_digest=evidence_digest, last_observed_at=_naive_utc(item.observed_at), status=( KnowledgeFSMigrationIssueStatus.RESOLVED if decision is KnowledgeFSShadowAuthorizationDecision.MATCH else KnowledgeFSMigrationIssueStatus.OPEN ), ) ) return False def _assert_shadow_completion_replay( self, ledger: KnowledgeFSWorkspaceCutoverLedger, payload: ShadowCompletionInput, report: ShadowCompletionReport, ) -> None: evidence = ( cast(dict[str, object], payload.traffic_zero_evidence) if payload.traffic_zero_evidence is not None else None ) if ( ledger.shadow_completed_at != _naive_utc(payload.completed_at) or ledger.shadow_evidence_digest != report.evidence_digest or ledger.shadow_observation_count != report.observation_count or ledger.shadow_traffic_zero is not payload.traffic_zero or ledger.shadow_traffic_zero_evidence != evidence or ledger.shadow_latest_observed_revision != report.latest_observed_revision or ledger.shadow_producer != payload.producer or ledger.shadow_completed_by_operator != payload.completed_by_operator or ledger.shadow_completed_by_account_id != str(payload.completed_by_account_id) or ledger.shadow_window_started_at != (_naive_utc(payload.window_started_at) if payload.window_started_at is not None else None) or ledger.shadow_window_ended_at != (_naive_utc(payload.window_ended_at) if payload.window_ended_at is not None else None) ): raise KnowledgeFSCutoverConflictError("Shadow was already completed with different evidence") @staticmethod def _shadow_decision(item: ShadowAuthorizationObservationInput) -> KnowledgeFSShadowAuthorizationDecision: if item.legacy_allowed is None: return KnowledgeFSShadowAuthorizationDecision.UNKNOWN if item.legacy_allowed == item.dify_allowed: return KnowledgeFSShadowAuthorizationDecision.MATCH if item.legacy_allowed and not item.dify_allowed: return KnowledgeFSShadowAuthorizationDecision.TIGHTENED return KnowledgeFSShadowAuthorizationDecision.EXPANDED @staticmethod def _require_ledger( repository: SQLAlchemyKnowledgeFSCutoverRepository, tenant_id: str ) -> KnowledgeFSWorkspaceCutoverLedger: ledger = repository.get_ledger(tenant_id=tenant_id) if ledger is None: raise KnowledgeFSCutoverNotFoundError("Workspace cutover ledger was not found") return ledger @staticmethod def _assert_phase( ledger: KnowledgeFSWorkspaceCutoverLedger, expected_phase: KnowledgeFSWorkspaceCutoverPhase, expected_cas_version: int, ) -> None: if ledger.phase is not expected_phase: raise KnowledgeFSCutoverConflictError( f"Expected cutover phase {expected_phase.value}, found {ledger.phase.value}" ) if ledger.cas_version != expected_cas_version: raise KnowledgeFSCutoverConflictError("Cutover ledger version changed") @staticmethod def _assert_no_open_gates( repository: SQLAlchemyKnowledgeFSCutoverRepository, ledger: KnowledgeFSWorkspaceCutoverLedger, ) -> None: if repository.count_open_issues(tenant_id=ledger.tenant_id, ledger_id=ledger.id) > 0: raise KnowledgeFSCutoverGateBlockedError("Open migration issues block phase advancement") if repository.count_unapproved_shadow_diffs(tenant_id=ledger.tenant_id, ledger_id=ledger.id) > 0: raise KnowledgeFSCutoverGateBlockedError("Unapproved shadow differences block phase advancement") def _assert_shadow_evidence_complete( self, ledger: KnowledgeFSWorkspaceCutoverLedger, *, required_revision: KnowledgeFSCutoverRevisionWatermark | None = None, ) -> None: if ( ledger.shadow_started_at is None or ledger.shadow_completed_at is None or ledger.shadow_evidence_digest is None or ledger.shadow_producer not in self._trusted_shadow_producers or ledger.shadow_completed_by_operator not in self._trusted_shadow_operators or ledger.shadow_completed_by_account_id is None ): raise KnowledgeFSCutoverGateBlockedError("Shadow evidence is not complete and trusted") if ledger.shadow_traffic_zero: if ledger.shadow_observation_count != 0 or ledger.shadow_traffic_zero_evidence is None: raise KnowledgeFSCutoverGateBlockedError("Traffic-zero shadow evidence is incomplete") return if ( ledger.shadow_observation_count <= 0 or ledger.shadow_window_started_at is None or ledger.shadow_window_ended_at is None or ledger.shadow_latest_observed_revision is None ): raise KnowledgeFSCutoverGateBlockedError("Shadow observation evidence is incomplete") applicable_revision = required_revision or ledger.final_revision_watermark or ledger.source_revision_watermark if not _watermark_at_least(ledger.shadow_latest_observed_revision, applicable_revision): raise KnowledgeFSCutoverGateBlockedError( "Shadow latest observed revision is below the applicable cutover watermark" ) def _assert_cutover_ready( self, repository: SQLAlchemyKnowledgeFSCutoverRepository, ledger: KnowledgeFSWorkspaceCutoverLedger, expected_cas_version: int, ) -> None: self._assert_phase(ledger, KnowledgeFSWorkspaceCutoverPhase.FROZEN, expected_cas_version) # Final delta is applied only after freeze. Revalidation therefore uses the # source watermark that was applicable to the frozen shadow gate. self._assert_shadow_evidence_complete(ledger, required_revision=ledger.source_revision_watermark) self._assert_no_open_gates(repository, ledger) self._assert_no_unresolved_quarantine(repository, ledger) if ledger.freeze_at is None: raise KnowledgeFSCutoverGateBlockedError("Workspace freeze evidence is missing") _assert_remote_freeze_evidence(ledger) if not ledger.legacy_dependency_ready or ledger.legacy_dependency_checked_at is None: raise KnowledgeFSCutoverGateBlockedError("Legacy snapshot/FK dependency dashboard is not ready") if ( ledger.final_revision_watermark is None or ledger.final_revision_watermark != ledger.applied_revision_watermark or ledger.final_task_watermark is None or ledger.final_task_watermark != ledger.applied_task_watermark ): raise KnowledgeFSCutoverGateBlockedError("Final delta watermarks are not fully applied") @staticmethod def _require_activation_control_space( session: Session, *, tenant_id: str, control_space_id: str | None = None, ) -> KnowledgeFSControlSpace: statement = sa.select(KnowledgeFSControlSpace).where( KnowledgeFSControlSpace.tenant_id == tenant_id, KnowledgeFSControlSpace.state == KnowledgeFSControlSpaceState.ACTIVE, KnowledgeFSControlSpace.knowledge_space_id.is_not(None), ) if control_space_id is not None: statement = statement.where(KnowledgeFSControlSpace.id == control_space_id) control_space = session.scalar( statement.order_by(KnowledgeFSControlSpace.created_at, KnowledgeFSControlSpace.id).limit(1) ) if control_space is None: raise KnowledgeFSCutoverGateBlockedError( "A tenant-owned active control-space is required for activation audit" ) return control_space @staticmethod def _assert_no_unresolved_quarantine( repository: SQLAlchemyKnowledgeFSCutoverRepository, ledger: KnowledgeFSWorkspaceCutoverLedger, ) -> None: if ( repository.count_unresolved_quarantine( tenant_id=ledger.tenant_id, ledger_id=ledger.id, source_kinds=_CUTOVER_QUARANTINE_KINDS, ) > 0 ): raise KnowledgeFSCutoverGateBlockedError("Unresolved migration quarantine blocks freeze and cutover") @staticmethod def _cas( repository: SQLAlchemyKnowledgeFSCutoverRepository, update: KnowledgeFSCutoverCASUpdate, ) -> None: if not repository.compare_and_set(update): raise KnowledgeFSCutoverConflictError("Cutover ledger changed during transition") def _classify_task(task: LegacyTaskInventoryInput) -> KnowledgeFSMigrationQuarantineDisposition: if task.subject_id is None or task.subject_id.startswith("dify-workspace:") or task.account_id is None: return KnowledgeFSMigrationQuarantineDisposition.ISOLATE if task.state in {"running", "paused"}: return KnowledgeFSMigrationQuarantineDisposition.WAIT_FOR_COMPLETION if task.state == "queued": return KnowledgeFSMigrationQuarantineDisposition.MIGRATABLE if task.state in {"completed", "failed", "canceled"}: return KnowledgeFSMigrationQuarantineDisposition.CANCEL return KnowledgeFSMigrationQuarantineDisposition.CANCEL def _shadow_observation_digest(item: ShadowAuthorizationObservationInput) -> str: canonical = json.dumps( item.model_dump(mode="json"), sort_keys=True, separators=(",", ":"), ).encode() return f"sha256:{hashlib.sha256(canonical).hexdigest()}" def _shadow_observation_model( *, ledger: KnowledgeFSWorkspaceCutoverLedger, item: ShadowAuthorizationObservationInput, decision: KnowledgeFSShadowAuthorizationDecision, digest: str, ) -> KnowledgeFSShadowAuthorizationObservation: return KnowledgeFSShadowAuthorizationObservation( tenant_id=ledger.tenant_id, ledger_id=ledger.id, diff_key=item.diff_key, producer=item.producer, control_space_id=str(item.control_space_id) if item.control_space_id else None, principal=item.principal, legacy_allowed=item.legacy_allowed, dify_allowed=False if item.legacy_allowed is None else item.dify_allowed, decision=decision, reason=item.reason, observed_revision=item.observed_revision.to_record(), observed_at=_naive_utc(item.observed_at), evidence_digest=digest, ) def _shadow_observation_set_digest( observations: Iterable[KnowledgeFSShadowAuthorizationObservation], ) -> str: canonical = "\n".join(sorted(item.evidence_digest for item in observations)).encode() return f"sha256:{hashlib.sha256(canonical).hexdigest()}" def _shadow_traffic_zero_digest(payload: ShadowCompletionInput) -> str: canonical = json.dumps( { "completed_at": payload.completed_at.isoformat(), "completed_by_account_id": str(payload.completed_by_account_id), "completed_by_operator": payload.completed_by_operator, "producer": payload.producer, "traffic_zero_evidence": payload.traffic_zero_evidence, }, sort_keys=True, separators=(",", ":"), ).encode() return f"sha256:{hashlib.sha256(canonical).hexdigest()}" def _revision_digest(revision: KnowledgeFSCutoverRevisionWatermark) -> str: canonical = json.dumps(revision, sort_keys=True, separators=(",", ":")).encode() return f"sha256:{hashlib.sha256(canonical).hexdigest()}" def _canonical_freeze_id( *, freeze_revision: int, namespace_id: str, source_revision_digest: str, source_task_watermark: int, ) -> str: canonical = json.dumps( { "freezeRevision": freeze_revision, "namespaceId": namespace_id, "sourceRevisionDigest": source_revision_digest, "sourceTaskWatermark": source_task_watermark, }, ensure_ascii=False, sort_keys=True, separators=(",", ":"), ).encode() return f"sha256:{hashlib.sha256(canonical).hexdigest()}" def _freeze_request( ledger: KnowledgeFSWorkspaceCutoverLedger, *, control_space_id: str, ) -> KnowledgeFSDifyIntegrationFreezeRequest: freeze_revision = max(ledger.cas_version + 1, (ledger.remote_freeze_revision or 0) + 1) source_revision_digest = _revision_digest(ledger.source_revision_watermark) freeze_id = _canonical_freeze_id( freeze_revision=freeze_revision, namespace_id=ledger.tenant_id, source_revision_digest=source_revision_digest, source_task_watermark=ledger.source_task_watermark, ) return KnowledgeFSDifyIntegrationFreezeRequest( namespace_id=ledger.tenant_id, control_space_id=control_space_id, freeze_id=freeze_id, freeze_revision=freeze_revision, source_revision_digest=source_revision_digest, source_task_watermark=ledger.source_task_watermark, ) def _assert_exact_freeze_ack( request: KnowledgeFSDifyIntegrationFreezeRequest, ack: KnowledgeFSDifyIntegrationFreezeAck, ) -> None: if ( ack.namespace_id != request.namespace_id or ack.freeze_id != request.freeze_id or ack.freeze_revision != request.freeze_revision or ack.source_revision_digest != request.source_revision_digest or ack.source_task_watermark != request.source_task_watermark ): raise KnowledgeFSCutoverGateBlockedError( "KnowledgeFS remote freeze acknowledgement did not exactly match the request" ) def _assert_remote_freeze_evidence(ledger: KnowledgeFSWorkspaceCutoverLedger) -> None: fields = ( ledger.remote_freeze_id, ledger.remote_freeze_revision, ledger.remote_freeze_digest, ledger.remote_freeze_task_watermark, ledger.remote_freeze_control_space_id, ledger.remote_freeze_frozen_at, ledger.remote_freeze_updated_at, ledger.remote_freeze_acknowledged_at, ledger.remote_freeze_applied, ledger.remote_freeze_replayed, ) if any(value is None for value in fields): raise KnowledgeFSCutoverGateBlockedError("Remote Workspace freeze evidence is incomplete") freeze_revision = cast(int, ledger.remote_freeze_revision) digest = cast(str, ledger.remote_freeze_digest) task_watermark = cast(int, ledger.remote_freeze_task_watermark) if ( digest != _revision_digest(ledger.source_revision_watermark) or task_watermark != ledger.source_task_watermark or ledger.remote_freeze_id != _canonical_freeze_id( freeze_revision=freeze_revision, namespace_id=ledger.tenant_id, source_revision_digest=digest, source_task_watermark=task_watermark, ) or ledger.remote_freeze_applied is ledger.remote_freeze_replayed ): raise KnowledgeFSCutoverGateBlockedError("Remote Workspace freeze evidence is inconsistent") def _activation_request( ledger: KnowledgeFSWorkspaceCutoverLedger, *, control_space_id: str, ) -> KnowledgeFSDifyIntegrationActivationRequest: final_revision = ledger.final_revision_watermark final_task_watermark = ledger.final_task_watermark if final_revision is None or final_task_watermark is None: raise KnowledgeFSCutoverGateBlockedError("Final delta evidence is missing") digest = _activation_source_revision_digest(final_revision, final_task_watermark) activation_revision = max( ledger.cas_version + 1, (ledger.remote_activation_revision or 0) + 1, ) activation_id = _canonical_activation_id( activation_revision=activation_revision, namespace_id=ledger.tenant_id, source_revision_digest=digest, ) return KnowledgeFSDifyIntegrationActivationRequest( namespace_id=ledger.tenant_id, control_space_id=control_space_id, activation_id=activation_id, activation_revision=activation_revision, source_revision_digest=digest, ) def _assert_exact_activation_ack( request: KnowledgeFSDifyIntegrationActivationRequest, ack: KnowledgeFSDifyIntegrationActivationAck, ) -> None: if ( ack.namespace_id != request.namespace_id or ack.activation_id != request.activation_id or ack.activation_revision != request.activation_revision or ack.source_revision_digest != request.source_revision_digest ): raise KnowledgeFSCutoverGateBlockedError( "KnowledgeFS remote activation acknowledgement did not exactly match the request" ) def _activation_source_revision_digest( final_revision: KnowledgeFSCutoverRevisionWatermark, final_task_watermark: int, ) -> str: canonical = json.dumps( { "final_revision_watermark": final_revision, "final_task_watermark": final_task_watermark, }, sort_keys=True, separators=(",", ":"), ).encode() return f"sha256:{hashlib.sha256(canonical).hexdigest()}" def _canonical_activation_id( *, activation_revision: int, namespace_id: str, source_revision_digest: str, ) -> str: activation_envelope = json.dumps( { "activationRevision": activation_revision, "namespaceId": namespace_id, "sourceRevisionDigest": source_revision_digest, }, ensure_ascii=False, sort_keys=True, separators=(",", ":"), ).encode() return f"sha256:{hashlib.sha256(activation_envelope).hexdigest()}" def knowledge_fs_remote_freeze_evidence_consistent(ledger: KnowledgeFSWorkspaceCutoverLedger) -> bool: """Return whether persisted KFS freeze evidence exactly binds the source watermarks.""" try: _assert_remote_freeze_evidence(ledger) except KnowledgeFSCutoverGateBlockedError: return False return True def knowledge_fs_remote_activation_evidence_consistent(ledger: KnowledgeFSWorkspaceCutoverLedger) -> bool: """Return whether persisted KFS activation evidence exactly binds the final delta.""" fields = ( ledger.remote_activation_id, ledger.remote_activation_revision, ledger.remote_activation_digest, ledger.remote_activation_control_space_id, ledger.remote_activation_activated_at, ledger.remote_activation_updated_at, ledger.remote_activation_acknowledged_at, ledger.remote_activation_applied, ledger.remote_activation_replayed, ) if any(value is None for value in fields): return False final_revision = ledger.final_revision_watermark final_task_watermark = ledger.final_task_watermark if final_revision is None or final_task_watermark is None: return False activation_revision = cast(int, ledger.remote_activation_revision) digest = _activation_source_revision_digest(final_revision, final_task_watermark) return ( ledger.remote_activation_digest == digest and ledger.remote_activation_id == _canonical_activation_id( activation_revision=activation_revision, namespace_id=ledger.tenant_id, source_revision_digest=digest, ) and ledger.remote_activation_control_space_id == ledger.remote_freeze_control_space_id and ledger.remote_activation_applied is not ledger.remote_activation_replayed ) def _latest_shadow_revision( observations: Iterable[KnowledgeFSShadowAuthorizationObservation], ) -> KnowledgeFSCutoverRevisionWatermark: values = tuple(observations) return { "membership_epoch": max(item.observed_revision["membership_epoch"] for item in values), "space_acl_epoch": max(item.observed_revision["space_acl_epoch"] for item in values), "external_access_epoch": max(item.observed_revision["external_access_epoch"] for item in values), "content_policy_revision": max(item.observed_revision["content_policy_revision"] for item in values), } def _watermark_at_least( candidate: KnowledgeFSCutoverRevisionWatermark, baseline: KnowledgeFSCutoverRevisionWatermark, ) -> bool: return ( candidate["membership_epoch"] >= baseline["membership_epoch"] and candidate["space_acl_epoch"] >= baseline["space_acl_epoch"] and candidate["external_access_epoch"] >= baseline["external_access_epoch"] and candidate["content_policy_revision"] >= baseline["content_policy_revision"] ) def _naive_utc(value: datetime) -> datetime: if value.tzinfo is None: return value return value.astimezone(UTC).replace(tzinfo=None) def _iso(value: datetime | None) -> str | None: return value.replace(tzinfo=UTC).isoformat() if value is not None else None __all__ = [ "CutoverBackfillReport", "CutoverInventoryReport", "CutoverRevisionWatermarkInput", "CutoverSmokeResultsInput", "FinalDeltaInput", "KnowledgeFSCutoverConflictError", "KnowledgeFSCutoverError", "KnowledgeFSCutoverGateBlockedError", "KnowledgeFSCutoverNotFoundError", "KnowledgeFSWorkspaceCutoverService", "LegacyDependencyDashboard", "LegacyDependencyInput", "QuarantineResolutionInput", "QuarantineResolutionReport", "ShadowAuthorizationObservationInput", "ShadowAuthorizationReport", "ShadowCompletionInput", "ShadowCompletionReport", "WorkspaceInventoryInput", "knowledge_fs_remote_activation_evidence_consistent", "knowledge_fs_remote_freeze_evidence_consistent", ]