diff --git a/backend/api/v1/__init__.py b/backend/api/v1/__init__.py
index 21a17a9..ecaa22d 100644
--- a/backend/api/v1/__init__.py
+++ b/backend/api/v1/__init__.py
@@ -28,6 +28,7 @@
     webhooks,
     workers,
     workflows,
+    workspaces,
 )
 
 v1_router = APIRouter(prefix="/api/v1")
@@ -53,6 +54,7 @@
 v1_router.include_router(skill_record.router)
 v1_router.include_router(webhooks.router)
 v1_router.include_router(workflows.router)
+v1_router.include_router(workspaces.router)
 v1_router.include_router(notifications.router)
 v1_router.include_router(workers.router)
 v1_router.include_router(dashboard.router)
diff --git a/backend/api/v1/workspaces.py b/backend/api/v1/workspaces.py
new file mode 100644
index 0000000..4da2d95
--- /dev/null
+++ b/backend/api/v1/workspaces.py
@@ -0,0 +1,282 @@
+"""Workspace / Project / WorkflowDraft / WorkflowVersion authoring endpoints."""
+
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from backend.database import get_db
+from backend.schemas import workflow_authoring as schemas
+from backend.schemas.common import ApiResponse
+from backend.services import workflow_authoring_service as authoring_service
+from backend.services import validation_run_service
+from backend.services.workflow_authoring_service import DraftRevisionConflictError
+from backend.services.validation_run_service import (
+    ValidationRunAlreadyConsumedError,
+    ValidationRunNotFoundError,
+    ValidationRunNotPassedError,
+    ValidationRunRequestError,
+    ValidationRunStaleError,
+)
+
+router = APIRouter(tags=["workflow-authoring"])
+
+
+@router.post("/workspaces", response_model=ApiResponse[schemas.WorkspaceRead], status_code=201)
+async def create_workspace(
+    body: schemas.WorkspaceCreate,
+    db: AsyncSession = Depends(get_db),
+) -> ApiResponse[schemas.WorkspaceRead]:
+    try:
+        workspace = await authoring_service.create_workspace(db, body)
+        await db.commit()
+    except IntegrityError as exc:
+        await db.rollback()
+        raise HTTPException(status_code=409, detail=f"Workspace slug {body.slug!r} already exists") from exc
+    await db.refresh(workspace)
+    return ApiResponse.ok(schemas.WorkspaceRead.model_validate(workspace))
+
+
+@router.get("/workspaces", response_model=ApiResponse[list[schemas.WorkspaceRead]])
+async def list_workspaces(db: AsyncSession = Depends(get_db)) -> ApiResponse[list[schemas.WorkspaceRead]]:
+    workspaces = await authoring_service.list_workspaces(db)
+    return ApiResponse.ok([schemas.WorkspaceRead.model_validate(w) for w in workspaces])
+
+
+@router.get("/workspaces/{workspace_id}", response_model=ApiResponse[schemas.WorkspaceRead])
+async def get_workspace(
+    workspace_id: str, db: AsyncSession = Depends(get_db)
+) -> ApiResponse[schemas.WorkspaceRead]:
+    workspace = await authoring_service.get_workspace(db, workspace_id)
+    if workspace is None:
+        raise HTTPException(status_code=404, detail="Workspace not found")
+    return ApiResponse.ok(schemas.WorkspaceRead.model_validate(workspace))
+
+
+@router.get(
+    "/workspaces/{workspace_id}/settings",
+    response_model=ApiResponse[schemas.WorkspaceSettingsRead],
+)
+async def get_workspace_settings(
+    workspace_id: str, db: AsyncSession = Depends(get_db)
+) -> ApiResponse[schemas.WorkspaceSettingsRead]:
+    settings = await authoring_service.get_workspace_settings(db, workspace_id)
+    if settings is None:
+        raise HTTPException(status_code=404, detail="Workspace not found")
+    return ApiResponse.ok(schemas.WorkspaceSettingsRead.model_validate(settings))
+
+
+@router.put(
+    "/workspaces/{workspace_id}/settings",
+    response_model=ApiResponse[schemas.WorkspaceSettingsRead],
+)
+async def update_workspace_settings(
+    workspace_id: str,
+    body: schemas.WorkspaceSettingsUpdate,
+    db: AsyncSession = Depends(get_db),
+) -> ApiResponse[schemas.WorkspaceSettingsRead]:
+    settings = await authoring_service.get_workspace_settings(db, workspace_id)
+    if settings is None:
+        raise HTTPException(status_code=404, detail="Workspace not found")
+    settings = await authoring_service.update_workspace_settings(db, settings, body)
+    await db.commit()
+    await db.refresh(settings)
+    return ApiResponse.ok(schemas.WorkspaceSettingsRead.model_validate(settings))
+
+
+@router.post(
+    "/workspaces/{workspace_id}/projects",
+    response_model=ApiResponse[schemas.ProjectRead],
+    status_code=201,
+)
+async def create_project(
+    workspace_id: str,
+    body: schemas.ProjectCreate,
+    db: AsyncSession = Depends(get_db),
+) -> ApiResponse[schemas.ProjectRead]:
+    workspace = await authoring_service.get_workspace(db, workspace_id)
+    if workspace is None:
+        raise HTTPException(status_code=404, detail="Workspace not found")
+    try:
+        project = await authoring_service.create_project(db, workspace, body)
+        await db.commit()
+    except IntegrityError as exc:
+        await db.rollback()
+        raise HTTPException(
+            status_code=409, detail=f"Project slug {body.slug!r} already exists in this workspace"
+        ) from exc
+    await db.refresh(project)
+    return ApiResponse.ok(schemas.ProjectRead.model_validate(project))
+
+
+@router.get(
+    "/workspaces/{workspace_id}/projects",
+    response_model=ApiResponse[list[schemas.ProjectRead]],
+)
+async def list_projects(
+    workspace_id: str, db: AsyncSession = Depends(get_db)
+) -> ApiResponse[list[schemas.ProjectRead]]:
+    workspace = await authoring_service.get_workspace(db, workspace_id)
+    if workspace is None:
+        raise HTTPException(status_code=404, detail="Workspace not found")
+    projects = await authoring_service.list_projects(db, workspace_id)
+    return ApiResponse.ok([schemas.ProjectRead.model_validate(p) for p in projects])
+
+
+@router.get(
+    "/workspaces/{workspace_id}/projects/{project_id}",
+    response_model=ApiResponse[schemas.ProjectRead],
+)
+async def get_project_in_workspace(
+    workspace_id: str, project_id: str, db: AsyncSession = Depends(get_db)
+) -> ApiResponse[schemas.ProjectRead]:
+    project = await authoring_service.get_project(db, project_id)
+    if project is None or project.workspace_id != workspace_id:
+        raise HTTPException(status_code=404, detail="Project not found")
+    return ApiResponse.ok(schemas.ProjectRead.model_validate(project))
+
+
+@router.post(
+    "/projects/{project_id}/drafts",
+    response_model=ApiResponse[schemas.WorkflowDraftRead],
+    status_code=201,
+)
+async def create_draft(
+    project_id: str,
+    body: schemas.WorkflowDraftCreate,
+    db: AsyncSession = Depends(get_db),
+) -> ApiResponse[schemas.WorkflowDraftRead]:
+    project = await authoring_service.get_project(db, project_id)
+    if project is None:
+        raise HTTPException(status_code=404, detail="Project not found")
+    draft = await authoring_service.create_draft(db, project, body)
+    await db.commit()
+    await db.refresh(draft)
+    return ApiResponse.ok(schemas.WorkflowDraftRead.model_validate(draft))
+
+
+@router.get("/drafts/{draft_id}", response_model=ApiResponse[schemas.WorkflowDraftRead])
+async def get_draft(draft_id: str, db: AsyncSession = Depends(get_db)) -> ApiResponse[schemas.WorkflowDraftRead]:
+    draft = await authoring_service.get_draft(db, draft_id)
+    if draft is None:
+        raise HTTPException(status_code=404, detail="Draft not found")
+    return ApiResponse.ok(schemas.WorkflowDraftRead.model_validate(draft))
+
+
+@router.put("/drafts/{draft_id}", response_model=ApiResponse[schemas.WorkflowDraftRead])
+async def update_draft(
+    draft_id: str,
+    body: schemas.WorkflowDraftUpdate,
+    db: AsyncSession = Depends(get_db),
+) -> ApiResponse[schemas.WorkflowDraftRead]:
+    draft = await authoring_service.get_draft(db, draft_id)
+    if draft is None:
+        raise HTTPException(status_code=404, detail="Draft not found")
+    try:
+        draft = await authoring_service.update_draft(db, draft, body)
+    except DraftRevisionConflictError as exc:
+        raise HTTPException(status_code=409, detail=str(exc)) from exc
+    await db.commit()
+    await db.refresh(draft)
+    return ApiResponse.ok(schemas.WorkflowDraftRead.model_validate(draft))
+
+
+@router.post(
+    "/drafts/{draft_id}/validation-runs",
+    response_model=ApiResponse[schemas.ValidationRunRead],
+    status_code=201,
+)
+async def create_validation_run(
+    draft_id: str,
+    body: schemas.ValidationRunCreate,
+    db: AsyncSession = Depends(get_db),
+) -> ApiResponse[schemas.ValidationRunRead]:
+    draft = await authoring_service.get_draft(db, draft_id)
+    if draft is None:
+        raise HTTPException(status_code=404, detail="Draft not found")
+    try:
+        validation_run = await validation_run_service.run_validation(
+            db, draft, mode=body.mode, expected_events=body.expected_events
+        )
+    except ValidationRunRequestError as exc:
+        raise HTTPException(status_code=422, detail=str(exc)) from exc
+    await db.commit()
+    await db.refresh(validation_run)
+    return ApiResponse.ok(schemas.ValidationRunRead.model_validate(validation_run))
+
+
+@router.get(
+    "/drafts/{draft_id}/validation-runs/{validation_run_id}",
+    response_model=ApiResponse[schemas.ValidationRunRead],
+)
+async def get_validation_run(
+    draft_id: str, validation_run_id: str, db: AsyncSession = Depends(get_db)
+) -> ApiResponse[schemas.ValidationRunRead]:
+    validation_run = await validation_run_service.get_validation_run(db, validation_run_id)
+    if validation_run is None or validation_run.draft_id != draft_id:
+        raise HTTPException(status_code=404, detail="Validation run not found")
+    return ApiResponse.ok(schemas.ValidationRunRead.model_validate(validation_run))
+
+
+@router.post("/drafts/{draft_id}/publish", response_model=ApiResponse[schemas.WorkflowVersionRead])
+async def publish_draft(
+    draft_id: str,
+    body: schemas.WorkflowDraftPublishRequest,
+    db: AsyncSession = Depends(get_db),
+) -> ApiResponse[schemas.WorkflowVersionRead]:
+    draft = await authoring_service.get_draft(db, draft_id)
+    if draft is None:
+        raise HTTPException(status_code=404, detail="Draft not found")
+    try:
+        version = await validation_run_service.publish_draft(
+            db,
+            draft,
+            validation_run_id=body.validation_run_id,
+            expected_revision=body.expected_revision,
+        )
+        await db.commit()
+    except DraftRevisionConflictError as exc:
+        await db.rollback()
+        raise HTTPException(status_code=409, detail=str(exc)) from exc
+    except ValidationRunNotFoundError as exc:
+        await db.rollback()
+        raise HTTPException(status_code=404, detail=str(exc)) from exc
+    except ValidationRunStaleError as exc:
+        await db.rollback()
+        raise HTTPException(status_code=409, detail=str(exc)) from exc
+    except ValidationRunNotPassedError as exc:
+        await db.rollback()
+        raise HTTPException(status_code=409, detail=str(exc)) from exc
+    except ValidationRunAlreadyConsumedError as exc:
+        await db.rollback()
+        raise HTTPException(status_code=409, detail=str(exc)) from exc
+    except IntegrityError as exc:
+        await db.rollback()
+        raise HTTPException(
+            status_code=409, detail="Validation run has already been published (concurrent publish)"
+        ) from exc
+    await db.refresh(version)
+    return ApiResponse.ok(schemas.WorkflowVersionRead.model_validate(version))
+
+
+@router.get(
+    "/projects/{project_id}/versions",
+    response_model=ApiResponse[list[schemas.WorkflowVersionRead]],
+)
+async def list_versions(
+    project_id: str, db: AsyncSession = Depends(get_db)
+) -> ApiResponse[list[schemas.WorkflowVersionRead]]:
+    project = await authoring_service.get_project(db, project_id)
+    if project is None:
+        raise HTTPException(status_code=404, detail="Project not found")
+    versions = await authoring_service.list_versions(db, project_id)
+    return ApiResponse.ok([schemas.WorkflowVersionRead.model_validate(v) for v in versions])
+
+
+@router.get("/versions/{version_id}", response_model=ApiResponse[schemas.WorkflowVersionRead])
+async def get_version(
+    version_id: str, db: AsyncSession = Depends(get_db)
+) -> ApiResponse[schemas.WorkflowVersionRead]:
+    version = await authoring_service.get_version(db, version_id)
+    if version is None:
+        raise HTTPException(status_code=404, detail="Workflow version not found")
+    return ApiResponse.ok(schemas.WorkflowVersionRead.model_validate(version))
diff --git a/backend/migrations/versions/r5s6t7u8v9w0_add_workflow_authoring_tables.py b/backend/migrations/versions/r5s6t7u8v9w0_add_workflow_authoring_tables.py
new file mode 100644
index 0000000..06847a2
--- /dev/null
+++ b/backend/migrations/versions/r5s6t7u8v9w0_add_workflow_authoring_tables.py
@@ -0,0 +1,148 @@
+"""add workflow authoring tables
+
+Revision ID: r5s6t7u8v9w0
+Revises: m2n3o4p5q6r7
+Create Date: 2026-07-13
+
+Persist the Workspace -> Project -> WorkflowDraft -> WorkflowVersion authoring
+closed loop, plus the ValidationRun publish gate that a WorkflowVersion must
+reference before it can be inserted.
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "r5s6t7u8v9w0"
+down_revision = "m2n3o4p5q6r7"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "workspaces",
+        sa.Column("id", sa.String(length=36), nullable=False),
+        sa.Column("name", sa.String(length=255), nullable=False),
+        sa.Column("slug", sa.String(length=255), nullable=False),
+        sa.Column("description", sa.String(length=2000), nullable=True),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index("ix_workspaces_slug", "workspaces", ["slug"], unique=True)
+
+    op.create_table(
+        "workspace_settings",
+        sa.Column("id", sa.String(length=36), nullable=False),
+        sa.Column("workspace_id", sa.String(length=36), nullable=False),
+        sa.Column("timezone", sa.String(length=64), nullable=False),
+        sa.Column("deterministic_simulation", sa.Boolean(), nullable=False),
+        sa.Column("max_items_per_run", sa.Integer(), nullable=False),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+        sa.ForeignKeyConstraint(["workspace_id"], ["workspaces.id"], ondelete="CASCADE"),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index(
+        "ix_workspace_settings_workspace_id",
+        "workspace_settings",
+        ["workspace_id"],
+        unique=True,
+    )
+
+    op.create_table(
+        "projects",
+        sa.Column("id", sa.String(length=36), nullable=False),
+        sa.Column("workspace_id", sa.String(length=36), nullable=False),
+        sa.Column("name", sa.String(length=255), nullable=False),
+        sa.Column("slug", sa.String(length=255), nullable=False),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+        sa.ForeignKeyConstraint(["workspace_id"], ["workspaces.id"], ondelete="CASCADE"),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint("workspace_id", "slug", name="uq_projects_workspace_slug"),
+    )
+    op.create_index("ix_projects_workspace_id", "projects", ["workspace_id"])
+
+    op.create_table(
+        "workflow_drafts",
+        sa.Column("id", sa.String(length=36), nullable=False),
+        sa.Column("project_id", sa.String(length=36), nullable=False),
+        sa.Column("name", sa.String(length=255), nullable=False),
+        sa.Column("revision", sa.Integer(), nullable=False),
+        sa.Column("snapshot", sa.JSON(), nullable=False),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+        sa.ForeignKeyConstraint(["project_id"], ["projects.id"], ondelete="CASCADE"),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index("ix_workflow_drafts_project_id", "workflow_drafts", ["project_id"])
+
+    op.create_table(
+        "validation_runs",
+        sa.Column("id", sa.String(length=36), nullable=False),
+        sa.Column("draft_id", sa.String(length=36), nullable=False),
+        sa.Column("draft_revision", sa.Integer(), nullable=False),
+        sa.Column("status", sa.String(length=50), nullable=False),
+        sa.Column("compile_valid", sa.Boolean(), nullable=False),
+        sa.Column("compile_errors", sa.JSON(), nullable=True),
+        sa.Column("conformance_mode", sa.String(length=50), nullable=False),
+        sa.Column("expected_events", sa.JSON(), nullable=True),
+        sa.Column("conformance_result", sa.JSON(), nullable=True),
+        sa.Column("runtime_passport", sa.JSON(), nullable=True),
+        sa.Column("run_id", sa.String(length=36), nullable=True),
+        sa.Column("failure_reason", sa.String(length=2000), nullable=True),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+        sa.ForeignKeyConstraint(["draft_id"], ["workflow_drafts.id"], ondelete="CASCADE"),
+        sa.ForeignKeyConstraint(["run_id"], ["workflow_runs.id"], ondelete="SET NULL"),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index("ix_validation_runs_draft_id", "validation_runs", ["draft_id"])
+
+    op.create_table(
+        "workflow_versions",
+        sa.Column("id", sa.String(length=36), nullable=False),
+        sa.Column("project_id", sa.String(length=36), nullable=False),
+        sa.Column("draft_id", sa.String(length=36), nullable=True),
+        sa.Column("version_number", sa.Integer(), nullable=False),
+        sa.Column("source_revision", sa.Integer(), nullable=False),
+        sa.Column("validation_run_id", sa.String(length=36), nullable=False),
+        sa.Column("snapshot", sa.JSON(), nullable=False),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+        sa.ForeignKeyConstraint(["project_id"], ["projects.id"], ondelete="CASCADE"),
+        sa.ForeignKeyConstraint(["draft_id"], ["workflow_drafts.id"], ondelete="SET NULL"),
+        sa.ForeignKeyConstraint(
+            ["validation_run_id"], ["validation_runs.id"], ondelete="RESTRICT"
+        ),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint(
+            "project_id", "version_number", name="uq_workflow_versions_project_number"
+        ),
+    )
+    op.create_index("ix_workflow_versions_project_id", "workflow_versions", ["project_id"])
+    op.create_index("ix_workflow_versions_draft_id", "workflow_versions", ["draft_id"])
+    op.create_index(
+        "ix_workflow_versions_validation_run_id",
+        "workflow_versions",
+        ["validation_run_id"],
+        unique=True,
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("ix_workflow_versions_validation_run_id", table_name="workflow_versions")
+    op.drop_index("ix_workflow_versions_draft_id", table_name="workflow_versions")
+    op.drop_index("ix_workflow_versions_project_id", table_name="workflow_versions")
+    op.drop_table("workflow_versions")
+    op.drop_index("ix_validation_runs_draft_id", table_name="validation_runs")
+    op.drop_table("validation_runs")
+    op.drop_index("ix_workflow_drafts_project_id", table_name="workflow_drafts")
+    op.drop_table("workflow_drafts")
+    op.drop_index("ix_projects_workspace_id", table_name="projects")
+    op.drop_table("projects")
+    op.drop_index("ix_workspace_settings_workspace_id", table_name="workspace_settings")
+    op.drop_table("workspace_settings")
+    op.drop_index("ix_workspaces_slug", table_name="workspaces")
+    op.drop_table("workspaces")
diff --git a/backend/models/__init__.py b/backend/models/__init__.py
index 5648d8a..715343c 100644
--- a/backend/models/__init__.py
+++ b/backend/models/__init__.py
@@ -20,7 +20,15 @@
 from backend.models.source_cursor import SourceCursor
 from backend.models.source_measurement import SourceMeasurement
 from backend.models.task import CollectionTask, TaskRun, TaskRunEvent
+from backend.models.validation_run import ValidationRun
 from backend.models.worker import WorkerNode
+from backend.models.workflow_authoring import (
+    Project,
+    Workspace,
+    WorkspaceSettings,
+    WorkflowDraft,
+    WorkflowVersion,
+)
 from backend.models.workflow_run import WorkflowRun, WorkflowRunEvent
 
 __all__ = [
@@ -54,4 +62,10 @@
     "WorkerNode",
     "WorkflowRun",
     "WorkflowRunEvent",
+    "Workspace",
+    "WorkspaceSettings",
+    "Project",
+    "WorkflowDraft",
+    "WorkflowVersion",
+    "ValidationRun",
 ]
diff --git a/backend/models/validation_run.py b/backend/models/validation_run.py
new file mode 100644
index 0000000..1dbc20d
--- /dev/null
+++ b/backend/models/validation_run.py
@@ -0,0 +1,30 @@
+from sqlalchemy import JSON, Boolean, ForeignKey, Integer, String
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from backend.models.base import TimestampMixin
+
+
+class ValidationRun(TimestampMixin):
+    """A single compile+conformance validation attempt gating draft publish."""
+
+    __tablename__ = "validation_runs"
+
+    draft_id: Mapped[str] = mapped_column(
+        String(36), ForeignKey("workflow_drafts.id", ondelete="CASCADE"), nullable=False, index=True
+    )
+    draft_revision: Mapped[int] = mapped_column(Integer, nullable=False)
+    status: Mapped[str] = mapped_column(String(50), nullable=False, default="pending")
+    compile_valid: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
+    compile_errors: Mapped[list | None] = mapped_column(JSON, nullable=True)
+    conformance_mode: Mapped[str] = mapped_column(String(50), nullable=False, default="passthrough")
+    expected_events: Mapped[list | None] = mapped_column(JSON, nullable=True)
+    conformance_result: Mapped[dict | None] = mapped_column(JSON, nullable=True)
+    runtime_passport: Mapped[dict | None] = mapped_column(JSON, nullable=True)
+    run_id: Mapped[str | None] = mapped_column(
+        String(36), ForeignKey("workflow_runs.id", ondelete="SET NULL"), nullable=True
+    )
+    failure_reason: Mapped[str | None] = mapped_column(String(2000), nullable=True)
+
+    version: Mapped["WorkflowVersion | None"] = relationship(
+        "WorkflowVersion", back_populates="validation_run", uselist=False
+    )
diff --git a/backend/models/workflow_authoring.py b/backend/models/workflow_authoring.py
new file mode 100644
index 0000000..466c57c
--- /dev/null
+++ b/backend/models/workflow_authoring.py
@@ -0,0 +1,122 @@
+from sqlalchemy import JSON, Boolean, ForeignKey, Integer, String, UniqueConstraint
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from backend.models.base import TimestampMixin
+
+
+class Workspace(TimestampMixin):
+    """A top-level container for Projects and their shared settings."""
+
+    __tablename__ = "workspaces"
+
+    name: Mapped[str] = mapped_column(String(255), nullable=False)
+    slug: Mapped[str] = mapped_column(String(255), nullable=False, unique=True, index=True)
+    description: Mapped[str | None] = mapped_column(String(2000), nullable=True)
+
+    settings: Mapped["WorkspaceSettings"] = relationship(
+        "WorkspaceSettings",
+        back_populates="workspace",
+        uselist=False,
+        cascade="all, delete-orphan",
+    )
+    projects: Mapped[list["Project"]] = relationship(
+        "Project",
+        back_populates="workspace",
+        cascade="all, delete-orphan",
+    )
+
+
+class WorkspaceSettings(TimestampMixin):
+    """Workspace-scoped defaults applied to new WorkflowDraft snapshots."""
+
+    __tablename__ = "workspace_settings"
+
+    workspace_id: Mapped[str] = mapped_column(
+        String(36),
+        ForeignKey("workspaces.id", ondelete="CASCADE"),
+        nullable=False,
+        unique=True,
+        index=True,
+    )
+    timezone: Mapped[str] = mapped_column(String(64), nullable=False, default="Asia/Shanghai")
+    deterministic_simulation: Mapped[bool] = mapped_column(
+        Boolean, nullable=False, default=True
+    )
+    max_items_per_run: Mapped[int] = mapped_column(Integer, nullable=False, default=20)
+
+    workspace: Mapped["Workspace"] = relationship("Workspace", back_populates="settings")
+
+
+class Project(TimestampMixin):
+    """A named unit of workflow authoring inside a Workspace."""
+
+    __tablename__ = "projects"
+    __table_args__ = (UniqueConstraint("workspace_id", "slug", name="uq_projects_workspace_slug"),)
+
+    workspace_id: Mapped[str] = mapped_column(
+        String(36), ForeignKey("workspaces.id", ondelete="CASCADE"), nullable=False, index=True
+    )
+    name: Mapped[str] = mapped_column(String(255), nullable=False)
+    slug: Mapped[str] = mapped_column(String(255), nullable=False)
+
+    workspace: Mapped["Workspace"] = relationship("Workspace", back_populates="projects")
+    drafts: Mapped[list["WorkflowDraft"]] = relationship(
+        "WorkflowDraft",
+        back_populates="project",
+        cascade="all, delete-orphan",
+    )
+    versions: Mapped[list["WorkflowVersion"]] = relationship(
+        "WorkflowVersion",
+        back_populates="project",
+        cascade="all, delete-orphan",
+        order_by="WorkflowVersion.version_number",
+    )
+
+
+class WorkflowDraft(TimestampMixin):
+    """A mutable, revision-guarded WorkflowProject snapshot under authoring."""
+
+    __tablename__ = "workflow_drafts"
+
+    project_id: Mapped[str] = mapped_column(
+        String(36), ForeignKey("projects.id", ondelete="CASCADE"), nullable=False, index=True
+    )
+    name: Mapped[str] = mapped_column(String(255), nullable=False)
+    revision: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
+    snapshot: Mapped[dict] = mapped_column(JSON, nullable=False, default=dict)
+
+    project: Mapped["Project"] = relationship("Project", back_populates="drafts")
+    versions: Mapped[list["WorkflowVersion"]] = relationship(
+        "WorkflowVersion",
+        back_populates="draft",
+    )
+
+
+class WorkflowVersion(TimestampMixin):
+    """An immutable, published WorkflowProject snapshot."""
+
+    __tablename__ = "workflow_versions"
+    __table_args__ = (
+        UniqueConstraint("project_id", "version_number", name="uq_workflow_versions_project_number"),
+    )
+
+    project_id: Mapped[str] = mapped_column(
+        String(36), ForeignKey("projects.id", ondelete="CASCADE"), nullable=False, index=True
+    )
+    draft_id: Mapped[str | None] = mapped_column(
+        String(36), ForeignKey("workflow_drafts.id", ondelete="SET NULL"), nullable=True, index=True
+    )
+    version_number: Mapped[int] = mapped_column(Integer, nullable=False)
+    source_revision: Mapped[int] = mapped_column(Integer, nullable=False)
+    validation_run_id: Mapped[str] = mapped_column(
+        String(36),
+        ForeignKey("validation_runs.id", ondelete="RESTRICT"),
+        nullable=False,
+        unique=True,
+        index=True,
+    )
+    snapshot: Mapped[dict] = mapped_column(JSON, nullable=False, default=dict)
+
+    project: Mapped["Project"] = relationship("Project", back_populates="versions")
+    draft: Mapped["WorkflowDraft | None"] = relationship("WorkflowDraft", back_populates="versions")
+    validation_run: Mapped["ValidationRun"] = relationship("ValidationRun", back_populates="version")
diff --git a/backend/schemas/workflow.py b/backend/schemas/workflow.py
index d9a34a8..ed0f074 100644
--- a/backend/schemas/workflow.py
+++ b/backend/schemas/workflow.py
@@ -239,7 +239,7 @@ class WorkflowDemandDraftRequest(BaseModel):
     locale: Optional[str] = None
 
 
-ExternalWorkflowRuntime = Literal["langgraph", "langchain"]
+ExternalWorkflowRuntime = Literal["langgraph", "langchain", "dify", "n8n"]
 
 
 class WorkflowExternalImportRequest(BaseModel):
diff --git a/backend/schemas/workflow_authoring.py b/backend/schemas/workflow_authoring.py
new file mode 100644
index 0000000..4ce090b
--- /dev/null
+++ b/backend/schemas/workflow_authoring.py
@@ -0,0 +1,153 @@
+from datetime import datetime
+from typing import Any, Literal, Optional
+
+from pydantic import BaseModel, Field
+
+from backend.schemas.common import UTCModel
+from backend.schemas.workflow import WorkflowProject
+from backend.workflow.conformance.contracts import (
+    ConformanceCaseResult,
+    ExpectedWorkflowRunEvent,
+    RuntimePassport,
+)
+
+
+class WorkspaceCreate(BaseModel):
+    name: str = Field(..., min_length=1, max_length=255)
+    slug: str = Field(..., min_length=1, max_length=255)
+    description: Optional[str] = None
+
+
+class WorkspaceUpdate(BaseModel):
+    name: Optional[str] = None
+    description: Optional[str] = None
+
+
+class WorkspaceRead(UTCModel):
+    id: str
+    name: str
+    slug: str
+    description: Optional[str] = None
+    created_at: datetime
+    updated_at: datetime
+
+    model_config = {"from_attributes": True}
+
+
+class WorkspaceSettingsUpdate(BaseModel):
+    timezone: Optional[str] = None
+    deterministic_simulation: Optional[bool] = None
+    max_items_per_run: Optional[int] = Field(None, gt=0)
+
+
+class WorkspaceSettingsRead(UTCModel):
+    id: str
+    workspace_id: str
+    timezone: str
+    deterministic_simulation: bool
+    max_items_per_run: int
+    created_at: datetime
+    updated_at: datetime
+
+    model_config = {"from_attributes": True}
+
+
+class ProjectCreate(BaseModel):
+    name: str = Field(..., min_length=1, max_length=255)
+    slug: str = Field(..., min_length=1, max_length=255)
+
+
+class ProjectRead(UTCModel):
+    id: str
+    workspace_id: str
+    name: str
+    slug: str
+    created_at: datetime
+    updated_at: datetime
+
+    model_config = {"from_attributes": True}
+
+
+class WorkflowDraftCreate(BaseModel):
+    name: str = Field(..., min_length=1, max_length=255)
+    snapshot: WorkflowProject
+
+
+class WorkflowDraftUpdate(BaseModel):
+    snapshot: WorkflowProject
+    expected_revision: int = Field(..., ge=1)
+
+
+class WorkflowDraftRead(UTCModel):
+    id: str
+    project_id: str
+    name: str
+    revision: int
+    snapshot: WorkflowProject
+    created_at: datetime
+    updated_at: datetime
+
+    model_config = {"from_attributes": True}
+
+
+class ValidationRunCreate(BaseModel):
+    mode: Literal["fixture", "passthrough"] = "passthrough"
+    expected_events: Optional[list[ExpectedWorkflowRunEvent]] = None
+
+
+class ValidationRunRead(UTCModel):
+    id: str
+    draft_id: str
+    draft_revision: int
+    status: str
+    compile_valid: bool
+    compile_errors: Optional[list[dict[str, Any]]] = None
+    conformance_mode: str
+    expected_events: Optional[list[dict[str, Any]]] = None
+    conformance_result: Optional[dict[str, Any]] = None
+    runtime_passport: Optional[dict[str, Any]] = None
+    run_id: Optional[str] = None
+    failure_reason: Optional[str] = None
+    created_at: datetime
+    updated_at: datetime
+
+    model_config = {"from_attributes": True}
+
+
+class WorkflowDraftPublishRequest(BaseModel):
+    validation_run_id: str = Field(..., min_length=1)
+    expected_revision: int = Field(..., ge=1)
+
+
+class WorkflowVersionRead(UTCModel):
+    id: str
+    project_id: str
+    draft_id: Optional[str] = None
+    version_number: int
+    source_revision: int
+    validation_run_id: str
+    snapshot: WorkflowProject
+    created_at: datetime
+    updated_at: datetime
+
+    model_config = {"from_attributes": True}
+
+
+__all__ = [
+    "WorkspaceCreate",
+    "WorkspaceUpdate",
+    "WorkspaceRead",
+    "WorkspaceSettingsUpdate",
+    "WorkspaceSettingsRead",
+    "ProjectCreate",
+    "ProjectRead",
+    "WorkflowDraftCreate",
+    "WorkflowDraftUpdate",
+    "WorkflowDraftRead",
+    "ValidationRunCreate",
+    "ValidationRunRead",
+    "WorkflowDraftPublishRequest",
+    "WorkflowVersionRead",
+    "ConformanceCaseResult",
+    "RuntimePassport",
+]
diff --git a/backend/services/validation_run_service.py b/backend/services/validation_run_service.py
new file mode 100644
index 0000000..3be9cd6
--- /dev/null
+++ b/backend/services/validation_run_service.py
@@ -0,0 +1,178 @@
+from typing import Literal, Optional
+
+from sqlalchemy import func, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from backend.models.validation_run import ValidationRun
+from backend.models.workflow_authoring import WorkflowDraft, WorkflowVersion
+from backend.schemas.workflow import WorkflowProject, WorkflowRunStartRequest
+from backend.services.workflow_authoring_service import DraftRevisionConflictError
+from backend.workflow.compiler import compile_workflow_project
+from backend.workflow.conformance.contracts import (
+    ConformanceCaseResult,
+    ExpectedWorkflowRunEvent,
+    build_runtime_passport,
+    match_expected_events,
+)
+from backend.workflow.opencli_hda_tracer import list_workflow_run_events, start_workflow_run
+
+
+class ValidationRunRequestError(ValueError):
+    """Raised when a validation run is requested with an invalid mode/payload combination."""
+
+
+class ValidationRunNotFoundError(ValueError):
+    """Raised when a publish references a validation run that doesn't belong to the draft."""
+
+
+class ValidationRunStaleError(ValueError):
+    """Raised when a validation run was captured against a since-superseded draft revision."""
+
+
+class ValidationRunNotPassedError(ValueError):
+    """Raised when a publish references a validation run that never reached status 'passed'."""
+
+
+class ValidationRunAlreadyConsumedError(ValueError):
+    """Raised when a publish references a validation run a prior WorkflowVersion already used."""
+
+
+async def run_validation(
+    session: AsyncSession,
+    draft: WorkflowDraft,
+    *,
+    mode: Literal["fixture", "passthrough"] = "passthrough",
+    expected_events: Optional[list[ExpectedWorkflowRunEvent]] = None,
+) -> ValidationRun:
+    if mode == "fixture" and not expected_events:
+        raise ValidationRunRequestError("fixture mode requires a non-empty expected_events list")
+
+    project = WorkflowProject.model_validate(draft.snapshot)
+    validation_run = ValidationRun(
+        draft_id=draft.id,
+        draft_revision=draft.revision,
+        status="pending",
+        compile_valid=False,
+        conformance_mode=mode,
+        expected_events=(
+            [event.model_dump(mode="json") for event in expected_events]
+            if expected_events
+            else None
+        ),
+    )
+    session.add(validation_run)
+    await session.flush()
+
+    compile_result = compile_workflow_project(project)
+    validation_run.compile_valid = compile_result.valid
+    validation_run.compile_errors = (
+        [error.model_dump(mode="json") for error in compile_result.errors]
+        if compile_result.errors
+        else None
+    )
+    if not compile_result.valid:
+        validation_run.status = "failed"
+        validation_run.failure_reason = "compile_failed"
+        await session.flush()
+        await session.refresh(validation_run)
+        return validation_run
+
+    projection = await start_workflow_run(
+        WorkflowRunStartRequest(project=project), session=session
+    )
+    validation_run.run_id = projection.runId
+
+    events = await list_workflow_run_events(projection.runId, session=session) or []
+    actual_events = [event.model_dump(mode="json") for event in events]
+
+    if mode == "fixture":
+        match_result = match_expected_events(actual_events, expected_events or [])
+        passed = match_result.passed
+        failures = match_result.failures
+        conformance_result: dict = match_result.model_dump(mode="json")
+    else:
+        failed_events = [event for event in actual_events if event.get("eventType") == "failed"]
+        failures = [
+            f"node {event.get('nodeId')} reported a failed event" for event in failed_events
+        ]
+        passed = projection.valid and projection.status != "failed" and not failed_events
+        conformance_result = {"passed": passed, "failures": failures}
+
+    case_result = ConformanceCaseResult(
+        id=draft.id,
+        status="passed" if passed else "failed",
+        failures=failures,
+    )
+    passport = build_runtime_passport([case_result])
+
+    validation_run.conformance_result = conformance_result
+    validation_run.runtime_passport = passport.model_dump(mode="json")
+    validation_run.status = "passed" if passed else "failed"
+    if not passed:
+        validation_run.failure_reason = "conformance_failed"
+
+    await session.flush()
+    await session.refresh(validation_run)
+    return validation_run
+
+
+async def get_validation_run(session: AsyncSession, validation_run_id: str) -> Optional[ValidationRun]:
+    result = await session.execute(
+        select(ValidationRun).where(ValidationRun.id == validation_run_id)
+    )
+    return result.scalar_one_or_none()
+
+
+async def publish_draft(
+    session: AsyncSession,
+    draft: WorkflowDraft,
+    *,
+    validation_run_id: str,
+    expected_revision: int,
+) -> WorkflowVersion:
+    if draft.revision != expected_revision:
+        raise DraftRevisionConflictError(
+            f"draft revision {draft.revision} does not match expected {expected_revision}"
+        )
+
+    validation_run = await get_validation_run(session, validation_run_id)
+    if validation_run is None or validation_run.draft_id != draft.id:
+        raise ValidationRunNotFoundError(
+            f"validation run {validation_run_id!r} not found for draft {draft.id!r}"
+        )
+    if validation_run.draft_revision != expected_revision:
+        raise ValidationRunStaleError(
+            "validation run was captured against a stale draft revision"
+        )
+    if validation_run.status != "passed":
+        raise ValidationRunNotPassedError(
+            f"validation run status is {validation_run.status!r}, expected 'passed'"
+        )
+
+    existing = await session.execute(
+        select(WorkflowVersion).where(WorkflowVersion.validation_run_id == validation_run_id)
+    )
+    if existing.scalar_one_or_none() is not None:
+        raise ValidationRunAlreadyConsumedError(
+            f"validation run {validation_run_id!r} has already been published"
+        )
+
+    next_version_result = await session.execute(
+        select(func.max(WorkflowVersion.version_number)).where(
+            WorkflowVersion.project_id == draft.project_id
+        )
+    )
+    next_version_number = (next_version_result.scalar_one_or_none() or 0) + 1
+
+    version = WorkflowVersion(
+        project_id=draft.project_id,
+        draft_id=draft.id,
+        version_number=next_version_number,
+        source_revision=draft.revision,
+        validation_run_id=validation_run.id,
+        snapshot=draft.snapshot,
+    )
+    session.add(version)
+    await session.flush()
+    await session.refresh(version)
+    return version
diff --git a/backend/services/workflow_authoring_service.py b/backend/services/workflow_authoring_service.py
new file mode 100644
index 0000000..65b14a7
--- /dev/null
+++ b/backend/services/workflow_authoring_service.py
@@ -0,0 +1,150 @@
+from typing import Optional
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from backend.models.workflow_authoring import (
+    Project,
+    Workspace,
+    WorkspaceSettings,
+    WorkflowDraft,
+    WorkflowVersion,
+)
+from backend.schemas.workflow_authoring import (
+    ProjectCreate,
+    WorkspaceCreate,
+    WorkspaceSettingsUpdate,
+    WorkspaceUpdate,
+    WorkflowDraftCreate,
+    WorkflowDraftUpdate,
+)
+
+
+class DraftRevisionConflictError(ValueError):
+    """Raised when a draft update's expectedRevision no longer matches the stored revision."""
+
+
+async def create_workspace(session: AsyncSession, data: WorkspaceCreate) -> Workspace:
+    workspace = Workspace(**data.model_dump())
+    session.add(workspace)
+    await session.flush()
+    session.add(WorkspaceSettings(workspace_id=workspace.id))
+    await session.flush()
+    await session.refresh(workspace)
+    return workspace
+
+
+async def list_workspaces(session: AsyncSession) -> list[Workspace]:
+    result = await session.execute(select(Workspace).order_by(Workspace.created_at.desc()))
+    return list(result.scalars().all())
+
+
+async def get_workspace(session: AsyncSession, workspace_id: str) -> Optional[Workspace]:
+    result = await session.execute(select(Workspace).where(Workspace.id == workspace_id))
+    return result.scalar_one_or_none()
+
+
+async def update_workspace(
+    session: AsyncSession, workspace: Workspace, data: WorkspaceUpdate
+) -> Workspace:
+    updates = data.model_dump(exclude_unset=True)
+    for key, value in updates.items():
+        setattr(workspace, key, value)
+    await session.flush()
+    await session.refresh(workspace)
+    return workspace
+
+
+async def get_workspace_settings(
+    session: AsyncSession, workspace_id: str
+) -> Optional[WorkspaceSettings]:
+    result = await session.execute(
+        select(WorkspaceSettings).where(WorkspaceSettings.workspace_id == workspace_id)
+    )
+    return result.scalar_one_or_none()
+
+
+async def update_workspace_settings(
+    session: AsyncSession, settings: WorkspaceSettings, data: WorkspaceSettingsUpdate
+) -> WorkspaceSettings:
+    updates = data.model_dump(exclude_unset=True)
+    for key, value in updates.items():
+        setattr(settings, key, value)
+    await session.flush()
+    await session.refresh(settings)
+    return settings
+
+
+async def create_project(
+    session: AsyncSession, workspace: Workspace, data: ProjectCreate
+) -> Project:
+    project = Project(workspace_id=workspace.id, **data.model_dump())
+    session.add(project)
+    await session.flush()
+    await session.refresh(project)
+    return project
+
+
+async def list_projects(session: AsyncSession, workspace_id: str) -> list[Project]:
+    result = await session.execute(
+        select(Project)
+        .where(Project.workspace_id == workspace_id)
+        .order_by(Project.created_at.desc())
+    )
+    return list(result.scalars().all())
+
+
+async def get_project(session: AsyncSession, project_id: str) -> Optional[Project]:
+    result = await session.execute(select(Project).where(Project.id == project_id))
+    return result.scalar_one_or_none()
+
+
+async def create_draft(
+    session: AsyncSession, project: Project, data: WorkflowDraftCreate
+) -> WorkflowDraft:
+    draft = WorkflowDraft(
+        project_id=project.id,
+        name=data.name,
+        revision=1,
+        snapshot=data.snapshot.model_dump(mode="json"),
+    )
+    session.add(draft)
+    await session.flush()
+    await session.refresh(draft)
+    return draft
+
+
+async def get_draft(session: AsyncSession, draft_id: str) -> Optional[WorkflowDraft]:
+    result = await session.execute(select(WorkflowDraft).where(WorkflowDraft.id == draft_id))
+    return result.scalar_one_or_none()
+
+
+async def update_draft(
+    session: AsyncSession, draft: WorkflowDraft, data: WorkflowDraftUpdate
+) -> WorkflowDraft:
+    if draft.revision != data.expected_revision:
+        raise DraftRevisionConflictError(
+            f"draft revision {draft.revision} does not match expected "
+            f"{data.expected_revision}"
+        )
+    draft.snapshot = data.snapshot.model_dump(mode="json")
+    draft.revision += 1
+    await session.flush()
+    await session.refresh(draft)
+    return draft
+
+
+async def list_versions(session: AsyncSession, project_id: str) -> list[WorkflowVersion]:
+    result = await session.execute(
+        select(WorkflowVersion)
+        .where(WorkflowVersion.project_id == project_id)
+        .order_by(WorkflowVersion.version_number.desc())
+    )
+    return list(result.scalars().all())
+
+
+async def get_version(session: AsyncSession, version_id: str) -> Optional[WorkflowVersion]:
+    result = await session.execute(
+        select(WorkflowVersion).where(WorkflowVersion.id == version_id)
+    )
+    return result.scalar_one_or_none()
diff --git a/backend/workflow/compiler.py b/backend/workflow/compiler.py
index 09b5a52..efb0f6f 100644
--- a/backend/workflow/compiler.py
+++ b/backend/workflow/compiler.py
@@ -27,6 +27,7 @@
 from backend.workflow.runtime_registry import resolve_runtime_metadata
 
 INTERNAL_ID_SEPARATOR = "::"
+MAX_PACKAGE_NESTING_DEPTH = 16
 
 
 @dataclass(frozen=True)
@@ -112,51 +113,13 @@ def compile_workflow_project(project: WorkflowProject) -> WorkflowCompileRespons
         plan_nodes.append(_to_plan_node(node))
 
         if node.internals:
-            bound_internal_nodes = _bind_internal_parameters(node)
-            internal_depends_on = _dependency_map_for_nodes(
-                bound_internal_nodes, node.internals.edges
+            nested_nodes, nested_plan_nodes, nested_edges, nested_plan_edges = (
+                _expand_package_internals(node, adapter_by_id, package_path=[node.id])
             )
-            locked = _package_locked(node)
-            for internal_node in bound_internal_nodes:
-                internal_id = _internal_id(node.id, internal_node.id)
-                internal_upstream = internal_depends_on[internal_node.id]
-                compiled_nodes.append(
-                    _compile_node(
-                        internal_node,
-                        adapter_by_id.get(internal_node.adapter or ""),
-                        [_internal_id(node.id, upstream) for upstream in internal_upstream]
-                        or [node.id],
-                        id_override=internal_id,
-                        runtime={
-                            "package_parent_id": node.id,
-                            "package_internal_id": internal_node.id,
-                            "editable": not locked,
-                        },
-                    )
-                )
-                plan_nodes.append(_to_plan_node(internal_node, id_override=internal_id))
-
-            for edge in node.internals.edges:
-                compiled_edges.append(
-                    CompiledWorkflowEdge(
-                        id=_internal_id(node.id, edge.id),
-                        source=_internal_id(node.id, edge.source),
-                        target=_internal_id(node.id, edge.target),
-                        sourcePort=edge.sourcePort or "records",
-                        targetPort=edge.targetPort or "records",
-                        contractId=edge.contractId,
-                        condition=edge.condition,
-                    )
-                )
-                plan_edges.append(
-                    PlanEdge(
-                        id=_internal_id(node.id, edge.id),
-                        source_node=_internal_id(node.id, edge.source),
-                        source_port=edge.sourcePort or "records",
-                        target_node=_internal_id(node.id, edge.target),
-                        target_port=edge.targetPort or "records",
-                    )
-                )
+            compiled_nodes.extend(nested_nodes)
+            plan_nodes.extend(nested_plan_nodes)
+            compiled_edges.extend(nested_edges)
+            plan_edges.extend(nested_plan_edges)
 
     compiled_edges = [
         CompiledWorkflowEdge(
@@ -279,7 +242,14 @@ def _validate_project(project: WorkflowProject) -> list[WorkflowCompileError]:
             )
 
         if node.internals:
-            errors.extend(_validate_package_internals(node, adapter_by_id))
+            errors.extend(
+                _validate_package_internals(
+                    node,
+                    adapter_by_id,
+                    package_path=[node.id],
+                    path_prefix=["nodes", node.id],
+                )
+            )
 
         errors.extend(_validate_node_origin(node, ["nodes", node.id]))
 
@@ -480,6 +450,11 @@ def _internal_id(package_node_id: str, internal_node_id: str) -> str:
     return f"{package_node_id}{INTERNAL_ID_SEPARATOR}{internal_node_id}"
 
 
+def _scoped_id(package_path: list[str], local_id: str | None = None) -> str:
+    parts = package_path if local_id is None else [*package_path, local_id]
+    return INTERNAL_ID_SEPARATOR.join(parts)
+
+
 def _package_locked(node: WorkflowProjectNode) -> bool:
     if node.internals and node.internals.locked is not None:
         return node.internals.locked
@@ -535,13 +510,100 @@ def _bind_internal_parameters(node: WorkflowProjectNode) -> list[WorkflowProject
     ]
 
 
+def _expand_package_internals(
+    node: WorkflowProjectNode,
+    adapter_by_id: dict[str, WorkflowAdapterBinding],
+    *,
+    package_path: list[str],
+) -> tuple[
+    list[CompiledWorkflowNode],
+    list[PlanNode],
+    list[CompiledWorkflowEdge],
+    list[PlanEdge],
+]:
+    compiled_nodes: list[CompiledWorkflowNode] = []
+    plan_nodes: list[PlanNode] = []
+    compiled_edges: list[CompiledWorkflowEdge] = []
+    plan_edges: list[PlanEdge] = []
+
+    if not node.internals:
+        return compiled_nodes, plan_nodes, compiled_edges, plan_edges
+
+    bound_internal_nodes = _bind_internal_parameters(node)
+    internal_depends_on = _dependency_map_for_nodes(bound_internal_nodes, node.internals.edges)
+    locked = _package_locked(node)
+    parent_scoped_id = _scoped_id(package_path)
+
+    for internal_node in bound_internal_nodes:
+        internal_id = _scoped_id(package_path, internal_node.id)
+        internal_upstream = internal_depends_on[internal_node.id]
+        compiled_nodes.append(
+            _compile_node(
+                internal_node,
+                adapter_by_id.get(internal_node.adapter or ""),
+                [_scoped_id(package_path, upstream) for upstream in internal_upstream]
+                or [parent_scoped_id],
+                id_override=internal_id,
+                runtime={
+                    "package_parent_id": parent_scoped_id,
+                    "package_internal_id": internal_node.id,
+                    "editable": not locked,
+                },
+            )
+        )
+        plan_nodes.append(_to_plan_node(internal_node, id_override=internal_id))
+
+        if internal_node.internals:
+            nested_nodes, nested_plan_nodes, nested_edges, nested_plan_edges = (
+                _expand_package_internals(
+                    internal_node,
+                    adapter_by_id,
+                    package_path=[*package_path, internal_node.id],
+                )
+            )
+            compiled_nodes.extend(nested_nodes)
+            plan_nodes.extend(nested_plan_nodes)
+            compiled_edges.extend(nested_edges)
+            plan_edges.extend(nested_plan_edges)
+
+    for edge in node.internals.edges:
+        compiled_edges.append(
+            CompiledWorkflowEdge(
+                id=_scoped_id(package_path, edge.id),
+                source=_scoped_id(package_path, edge.source),
+                target=_scoped_id(package_path, edge.target),
+                sourcePort=edge.sourcePort or "records",
+                targetPort=edge.targetPort or "records",
+                contractId=edge.contractId,
+                condition=edge.condition,
+            )
+        )
+        plan_edges.append(
+            PlanEdge(
+                id=_scoped_id(package_path, edge.id),
+                source_node=_scoped_id(package_path, edge.source),
+                source_port=edge.sourcePort or "records",
+                target_node=_scoped_id(package_path, edge.target),
+                target_port=edge.targetPort or "records",
+            )
+        )
+
+    return compiled_nodes, plan_nodes, compiled_edges, plan_edges
+
+
 def _validate_package_internals(
     node: WorkflowProjectNode,
     adapter_by_id: dict[str, WorkflowAdapterBinding],
+    *,
+    package_path: list[str],
+    path_prefix: list[str],
+    depth: int = 1,
 ) -> list[WorkflowCompileError]:
     errors: list[WorkflowCompileError] = []
     assert node.internals is not None
 
+    scoped_id = _scoped_id(package_path)
+
     internal_counts = Counter(internal_node.id for internal_node in node.internals.nodes)
     for internal_node_id, count in sorted(internal_counts.items()):
         if count > 1:
@@ -549,11 +611,11 @@ def _validate_package_internals(
                 WorkflowCompileError(
                     code="duplicate_internal_node_id",
                     message=(
-                        f'Package node "{node.id}" has duplicated internal node '
+                        f'Package node "{scoped_id}" has duplicated internal node '
                         f'"{internal_node_id}"'
                     ),
-                    node_id=node.id,
-                    path=["nodes", node.id, "internals", "nodes", internal_node_id],
+                    node_id=scoped_id,
+                    path=[*path_prefix, "internals", "nodes", internal_node_id],
                 )
             )
 
@@ -564,12 +626,12 @@ def _validate_package_internals(
                 WorkflowCompileError(
                     code="missing_internal_edge_source",
                     message=(
-                        f'Package node "{node.id}" internal edge "{edge.id}" '
+                        f'Package node "{scoped_id}" internal edge "{edge.id}" '
                         f'references missing source "{edge.source}"'
                     ),
-                    node_id=node.id,
+                    node_id=scoped_id,
                     edge_id=edge.id,
-                    path=["nodes", node.id, "internals", "edges", edge.id, "source"],
+                    path=[*path_prefix, "internals", "edges", edge.id, "source"],
                 )
             )
         if edge.target not in internal_node_ids:
@@ -577,12 +639,12 @@ def _validate_package_internals(
                 WorkflowCompileError(
                     code="missing_internal_edge_target",
                     message=(
-                        f'Package node "{node.id}" internal edge "{edge.id}" '
+                        f'Package node "{scoped_id}" internal edge "{edge.id}" '
                         f'references missing target "{edge.target}"'
                     ),
-                    node_id=node.id,
+                    node_id=scoped_id,
                     edge_id=edge.id,
-                    path=["nodes", node.id, "internals", "edges", edge.id, "target"],
+                    path=[*path_prefix, "internals", "edges", edge.id, "target"],
                 )
             )
 
@@ -592,14 +654,13 @@ def _validate_package_internals(
                 WorkflowCompileError(
                     code="missing_adapter_binding",
                     message=(
-                        f'Package node "{node.id}" internal node '
+                        f'Package node "{scoped_id}" internal node '
                         f'"{internal_node.id}" references missing adapter '
                         f'"{internal_node.adapter}"'
                     ),
-                    node_id=node.id,
+                    node_id=scoped_id,
                     path=[
-                        "nodes",
-                        node.id,
+                        *path_prefix,
                         "internals",
                         "nodes",
                         internal_node.id,
@@ -612,13 +673,12 @@ def _validate_package_internals(
                 WorkflowCompileError(
                     code="missing_adapter_binding",
                     message=(
-                        f'Package node "{node.id}" internal node '
+                        f'Package node "{scoped_id}" internal node '
                         f'"{internal_node.id}" requires an adapter binding'
                     ),
-                    node_id=node.id,
+                    node_id=scoped_id,
                     path=[
-                        "nodes",
-                        node.id,
+                        *path_prefix,
                         "internals",
                         "nodes",
                         internal_node.id,
@@ -630,10 +690,37 @@ def _validate_package_internals(
         errors.extend(
             _validate_node_origin(
                 internal_node,
-                ["nodes", node.id, "internals", "nodes", internal_node.id],
+                [*path_prefix, "internals", "nodes", internal_node.id],
             )
         )
 
+        if internal_node.internals:
+            child_package_path = [*package_path, internal_node.id]
+            child_path_prefix = [*path_prefix, "internals", "nodes", internal_node.id]
+            if depth >= MAX_PACKAGE_NESTING_DEPTH:
+                errors.append(
+                    WorkflowCompileError(
+                        code="package_nesting_limit_exceeded",
+                        message=(
+                            f'Package node "{_scoped_id(child_package_path)}" exceeds '
+                            f"the maximum {MAX_PACKAGE_NESTING_DEPTH}-level package "
+                            "nesting limit"
+                        ),
+                        node_id=_scoped_id(child_package_path),
+                        path=[*child_path_prefix, "internals"],
+                    )
+                )
+            else:
+                errors.extend(
+                    _validate_package_internals(
+                        internal_node,
+                        adapter_by_id,
+                        package_path=child_package_path,
+                        path_prefix=child_path_prefix,
+                        depth=depth + 1,
+                    )
+                )
+
     if node.parameterInterface:
         for field in node.parameterInterface.fields:
             if field.binding.nodeId not in internal_node_ids:
@@ -641,14 +728,13 @@ def _validate_package_internals(
                     WorkflowCompileError(
                         code="invalid_parameter_binding",
                         message=(
-                            f'Package node "{node.id}" public parameter '
+                            f'Package node "{scoped_id}" public parameter '
                             f'"{field.id}" binds missing internal node '
                             f'"{field.binding.nodeId}"'
                         ),
-                        node_id=node.id,
+                        node_id=scoped_id,
                         path=[
-                            "nodes",
-                            node.id,
+                            *path_prefix,
                             "parameterInterface",
                             "fields",
                             field.id,
@@ -661,15 +747,20 @@ def _validate_package_internals(
         _validate_typed_edges(
             node.internals.nodes,
             node.internals.edges,
-            path_prefix=["nodes", node.id, "internals", "edges"],
+            path_prefix=[*path_prefix, "internals", "edges"],
+        )
+    )
+    errors.extend(
+        _cycle_errors_for_nodes(
+            package_path, path_prefix, node.internals.nodes, node.internals.edges
         )
     )
-    errors.extend(_cycle_errors_for_nodes(node.id, node.internals.nodes, node.internals.edges))
     return errors
 
 
 def _cycle_errors_for_nodes(
-    package_node_id: str,
+    package_path: list[str],
+    path_prefix: list[str],
     nodes: list[WorkflowProjectNode],
     edges: list,
 ) -> list[WorkflowCompileError]:
@@ -691,15 +782,16 @@ def _cycle_errors_for_nodes(
             if indegree[target] == 0:
                 queue.append(target)
 
+    scoped_id = _scoped_id(package_path)
     return [
         WorkflowCompileError(
             code="cycle",
             message=(
-                f'Package node "{package_node_id}" internal graph contains '
+                f'Package node "{scoped_id}" internal graph contains '
                 f'a cycle at node "{node_id}"'
             ),
-            node_id=package_node_id,
-            path=["nodes", package_node_id, "internals", "nodes", node_id],
+            node_id=scoped_id,
+            path=[*path_prefix, "internals", "nodes", node_id],
         )
         for node_id in sorted(set(node_ids) - visited)
     ]
diff --git a/backend/workflow/conformance/contracts.py b/backend/workflow/conformance/contracts.py
index 2f7ad6b..fe1092c 100644
--- a/backend/workflow/conformance/contracts.py
+++ b/backend/workflow/conformance/contracts.py
@@ -120,20 +120,28 @@ def match_expected_events(
     )
 
 
-def write_runtime_passport(
-    artifact_dir: Path,
+def build_runtime_passport(
     case_results: list[ConformanceCaseResult],
     *,
     status: Literal["conformant", "partial", "preview-only", "failed"] | None = None,
-) -> Path:
-    artifact_dir.mkdir(parents=True, exist_ok=True)
-    passport = RuntimePassport(
+) -> RuntimePassport:
+    return RuntimePassport(
         generatedAt=datetime.now(UTC).isoformat().replace("+00:00", "Z"),
         status=status
         or ("failed" if any(case.status == "failed" for case in case_results) else "partial"),
         cases=case_results,
         bindings=_binding_evidence(case_results),
     )
+
+
+def write_runtime_passport(
+    artifact_dir: Path,
+    case_results: list[ConformanceCaseResult],
+    *,
+    status: Literal["conformant", "partial", "preview-only", "failed"] | None = None,
+) -> Path:
+    artifact_dir.mkdir(parents=True, exist_ok=True)
+    passport = build_runtime_passport(case_results, status=status)
     path = artifact_dir / "opencli-runtime-passport.json"
     path.write_text(
         json.dumps(passport.model_dump(mode="json"), indent=2, sort_keys=True) + "\n",
diff --git a/backend/workflow/external_importer.py b/backend/workflow/external_importer.py
index 443585e..6700298 100644
--- a/backend/workflow/external_importer.py
+++ b/backend/workflow/external_importer.py
@@ -24,8 +24,8 @@ def import_external_workflow(body: WorkflowExternalImportRequest) -> WorkflowPat
     OpenCLI Admin catalog capabilities and never carry external executors.
     """
 
-    external_nodes = _extract_nodes(body.graph)
-    external_edges = _extract_edges(body.graph)
+    external_nodes = _extract_nodes(body.graph, body.runtime)
+    external_edges = _extract_edges(body.graph, body.runtime)
     for edge in external_edges:
         external_nodes.setdefault(edge["source"], {"id": edge["source"]})
         external_nodes.setdefault(edge["target"], {"id": edge["target"]})
@@ -36,8 +36,9 @@ def import_external_workflow(body: WorkflowExternalImportRequest) -> WorkflowPat
     imported_node_ids: dict[str, str] = {}
     merge_input_counts: dict[str, int] = {}
 
+    prefer_name = body.runtime == "n8n"
     for index, external_node in enumerate(external_nodes.values()):
-        external_id = _node_id(external_node)
+        external_id = _node_id(external_node, prefer_name=prefer_name)
         node_id = _unique_id(used_node_ids, _slug(external_id))
         imported_node_ids[external_id] = node_id
         operations.append(
@@ -95,11 +96,25 @@ def _to_workflow_node(
     graph_name: str | None,
     index: int,
 ) -> WorkflowProjectNode:
-    external_id = _node_id(external_node)
+    external_id = _node_id(external_node, prefer_name=(runtime == "n8n"))
     external_type = _node_type(external_node)
     catalog_id, kind, capability, params = _native_capability_for_external_node(
         external_type
     )
+    ui: dict[str, Any] = {
+        "catalogId": catalog_id,
+        "label": _node_label(external_node),
+        "position": {"x": 180 + (index % 4) * 260, "y": 180 + (index // 4) * 140},
+        "externalWorkflow": {
+            "runtime": runtime,
+            "graphName": graph_name,
+            "nodeId": external_id,
+            "nodeType": external_type,
+            "raw": _safe_external_snapshot(external_node),
+        },
+    }
+    if runtime == "n8n":
+        ui["n8n"] = {"source": "n8n", "nodeId": external_id, "nodeType": external_type}
     return WorkflowProjectNode(
         id=node_id,
         kind=kind,
@@ -114,18 +129,7 @@ def _to_workflow_node(
                 "nodeType": external_type,
             },
         },
-        ui={
-            "catalogId": catalog_id,
-            "label": _node_label(external_node),
-            "position": {"x": 180 + (index % 4) * 260, "y": 180 + (index // 4) * 140},
-            "externalWorkflow": {
-                "runtime": runtime,
-                "graphName": graph_name,
-                "nodeId": external_id,
-                "nodeType": external_type,
-                "raw": _safe_external_snapshot(external_node),
-            },
-        },
+        ui=ui,
     )
 
 
@@ -174,24 +178,32 @@ def _native_capability_for_external_node(
     )
 
 
-def _extract_nodes(graph: dict[str, Any]) -> dict[str, dict[str, Any]]:
+def _extract_nodes(
+    graph: dict[str, Any], runtime: str | None = None
+) -> dict[str, dict[str, Any]]:
     raw_nodes = graph.get("nodes") or graph.get("vertices") or []
+    prefer_name = runtime == "n8n"
     nodes: dict[str, dict[str, Any]] = {}
     if isinstance(raw_nodes, dict):
         for key, value in raw_nodes.items():
             node = dict(value) if isinstance(value, dict) else {"value": value}
             node.setdefault("id", str(key))
-            nodes[_node_id(node)] = node
+            nodes[_node_id(node, prefer_name=prefer_name)] = node
         return nodes
     if isinstance(raw_nodes, list):
         for index, value in enumerate(raw_nodes):
             node = dict(value) if isinstance(value, dict) else {"value": value}
             node.setdefault("id", str(index))
-            nodes[_node_id(node)] = node
+            nodes[_node_id(node, prefer_name=prefer_name)] = node
     return nodes
 
 
-def _extract_edges(graph: dict[str, Any]) -> list[dict[str, Any]]:
+def _extract_edges(graph: dict[str, Any], runtime: str | None = None) -> list[dict[str, Any]]:
+    if runtime == "n8n":
+        connections = graph.get("connections")
+        if isinstance(connections, dict):
+            return _extract_n8n_connection_edges(connections)
+
     raw_edges = graph.get("edges") or graph.get("links") or []
     edges: list[dict[str, Any]] = []
     if not isinstance(raw_edges, list):
@@ -219,6 +231,42 @@ def _extract_edges(graph: dict[str, Any]) -> list[dict[str, Any]]:
     return edges
 
 
+def _extract_n8n_connection_edges(connections: dict[str, Any]) -> list[dict[str, Any]]:
+    """Flatten n8n's ``{sourceName: {main: [[{node, type, index}, ...]]}}`` shape.
+
+    n8n's export keys ``connections`` by node *name* (not the node's ``id``
+    field), and each output port fans out to a list of connection chains —
+    unlike the list-of-{source,target} shape every other runtime exports.
+    """
+
+    edges: list[dict[str, Any]] = []
+    for source_name, outputs in connections.items():
+        source = _read_string(source_name)
+        if not source or not isinstance(outputs, dict):
+            continue
+        for port_name, chains in outputs.items():
+            if not isinstance(chains, list):
+                continue
+            for chain in chains:
+                if not isinstance(chain, list):
+                    continue
+                for connection in chain:
+                    if not isinstance(connection, dict):
+                        continue
+                    target = _read_string(connection.get("node"))
+                    if not target:
+                        continue
+                    edges.append(
+                        {
+                            "id": f"edge-{len(edges) + 1}",
+                            "source": source,
+                            "target": target,
+                            "sourcePort": _read_string(port_name) or "main",
+                        }
+                    )
+    return edges
+
+
 def _operation_node_index(operations: list[WorkflowPatchOperation], node_id: str) -> int:
     for index, operation in enumerate(operations):
         if operation.node and operation.node.id == node_id:
@@ -247,7 +295,14 @@ def _target_port(
     return "in"
 
 
-def _node_id(node: dict[str, Any]) -> str:
+def _node_id(node: dict[str, Any], *, prefer_name: bool = False) -> str:
+    if prefer_name:
+        return (
+            _read_string(node.get("name"))
+            or _read_string(node.get("id"))
+            or _read_string(node.get("key"))
+            or "external-node"
+        )
     return (
         _read_string(node.get("id"))
         or _read_string(node.get("key"))
diff --git a/frontend/components/studio/workflow-lifecycle-strip.logic.ts b/frontend/components/studio/workflow-lifecycle-strip.logic.ts
new file mode 100644
index 0000000..26c837d
--- /dev/null
+++ b/frontend/components/studio/workflow-lifecycle-strip.logic.ts
@@ -0,0 +1,89 @@
+/**
+ * Pure state -> view derivation for WorkflowLifecycleStrip.
+ * Kept dependency-free (no JSX) so it can be imported directly by a
+ * plain Node regression script without a bundler/test runner.
+ */
+
+export type WorkflowLifecycleState =
+  | 'draft'
+  | 'validating'
+  | 'validated'
+  | 'publishing'
+  | 'published'
+  | 'blocked'
+
+export type LifecycleStageKey = 'draft' | 'validate' | 'publish' | 'activate'
+
+export type LifecycleStageStatus = 'pending' | 'active' | 'done' | 'error' | 'unavailable'
+
+export interface LifecycleStageView {
+  key: LifecycleStageKey
+  label: string
+  status: LifecycleStageStatus
+  note?: string
+}
+
+export interface WorkflowLifecycleView {
+  state: WorkflowLifecycleState
+  primaryStatusLabel: string
+  blockerText?: string
+  stages: LifecycleStageView[]
+}
+
+const STAGE_ORDER: LifecycleStageKey[] = ['draft', 'validate', 'publish', 'activate']
+
+const STAGE_LABELS: Record<LifecycleStageKey, string> = {
+  draft: '草稿',
+  validate: '验证',
+  publish: '版本发布',
+  activate: '激活',
+}
+
+const PRIMARY_STATUS_LABELS: Record<WorkflowLifecycleState, string> = {
+  draft: '草稿',
+  validating: '验证中',
+  validated: '已验证',
+  publishing: '发布中',
+  published: '已发布',
+  blocked: '已阻塞',
+}
+
+/**
+ * Activation has no backend-integrated meaning yet. This note must render
+ * unconditionally regardless of state — publishing/published must never
+ * imply the project is active.
+ */
+const ACTIVATION_NOTE = '待后端接入'
+
+const STAGE_STATUSES_BY_STATE: Record<
+  WorkflowLifecycleState,
+  Record<LifecycleStageKey, LifecycleStageStatus>
+> = {
+  draft: { draft: 'active', validate: 'pending', publish: 'pending', activate: 'unavailable' },
+  validating: { draft: 'done', validate: 'active', publish: 'pending', activate: 'unavailable' },
+  validated: { draft: 'done', validate: 'done', publish: 'pending', activate: 'unavailable' },
+  publishing: { draft: 'done', validate: 'done', publish: 'active', activate: 'unavailable' },
+  published: { draft: 'done', validate: 'done', publish: 'done', activate: 'unavailable' },
+  blocked: { draft: 'done', validate: 'error', publish: 'pending', activate: 'unavailable' },
+}
+
+export function deriveWorkflowLifecycleView(
+  state: WorkflowLifecycleState,
+  blockerText?: string,
+): WorkflowLifecycleView {
+  const statuses = STAGE_STATUSES_BY_STATE[state]
+
+  const stages: LifecycleStageView[] = STAGE_ORDER.map((key) => ({
+    key,
+    label: STAGE_LABELS[key],
+    status: statuses[key],
+    note: key === 'activate' ? ACTIVATION_NOTE : undefined,
+  }))
+
+  return {
+    state,
+    primaryStatusLabel: PRIMARY_STATUS_LABELS[state],
+    blockerText: state === 'blocked' ? blockerText : undefined,
+    stages,
+  }
+}
diff --git a/frontend/components/studio/workflow-lifecycle-strip.tsx b/frontend/components/studio/workflow-lifecycle-strip.tsx
new file mode 100644
index 0000000..ce58a13
--- /dev/null
+++ b/frontend/components/studio/workflow-lifecycle-strip.tsx
@@ -0,0 +1,78 @@
+import { Badge } from '@/components/ui/badge'
+import { cn } from '@/lib/utils'
+
+import {
+  deriveWorkflowLifecycleView,
+  type LifecycleStageStatus,
+  type WorkflowLifecycleState,
+} from './workflow-lifecycle-strip.logic'
+
+export interface WorkflowLifecycleStripProps {
+  /** Current authoring state of the workflow project. */
+  state: WorkflowLifecycleState
+  /** Concise reason the lifecycle is blocked. Only shown when state is "blocked". */
+  blockerText?: string
+  className?: string
+}
+
+const STAGE_STATUS_STYLES: Record<LifecycleStageStatus, string> = {
+  pending: 'border-border text-muted-foreground',
+  active: 'border-primary bg-primary text-primary-foreground',
+  done: 'border-success/40 bg-success/10 text-success',
+  error: 'border-destructive/40 bg-destructive/10 text-destructive',
+  unavailable: 'border-border border-dashed text-muted-foreground',
+}
+
+/**
+ * Compact strip of the four real authoring stages (draft/validate/publish)
+ * plus an always-unavailable activation stage — this repo has no deployment
+ * concept yet, so activation must never read as reachable or active.
+ */
+export function WorkflowLifecycleStrip({ state, blockerText, className }: WorkflowLifecycleStripProps) {
+  const view = deriveWorkflowLifecycleView(state, blockerText)
+
+  return (
+    <div
+      className={cn('flex flex-col gap-1.5', className)}
+      role="group"
+      aria-label={`工作流生命周期状态: ${view.primaryStatusLabel}`}
+    >
+      <div className="flex items-center gap-2">
+        <ol className="flex flex-1 items-center gap-1.5" role="list">
+          {view.stages.map((stage, index) => (
+            <li key={stage.key} className="flex flex-1 items-center gap-1.5" role="listitem">
+              <span
+                aria-current={stage.status === 'active' ? 'step' : undefined}
+                aria-label={
+                  stage.note ? `${stage.label}: ${stage.note}` : stage.label
+                }
+                className={cn(
+                  'flex h-6 flex-1 items-center justify-center gap-1 rounded-md border px-2 text-xs font-medium whitespace-nowrap',
+                  STAGE_STATUS_STYLES[stage.status],
+                )}
+              >
+                <span>{stage.label}</span>
+                {stage.note ? (
+                  <span className="text-muted-foreground" aria-hidden="true">
+                    ({stage.note})
+                  </span>
+                ) : null}
+              </span>
+              {index < view.stages.length - 1 ? (
+                <span aria-hidden="true" className="bg-border h-px w-2 shrink-0" />
+              ) : null}
+            </li>
+          ))}
+        </ol>
+        <Badge variant="outline" className="shrink-0">
+          {view.primaryStatusLabel}
+        </Badge>
+      </div>
+      {view.blockerText ? (
+        <p role="alert" className="text-destructive text-xs">
+          {view.blockerText}
+        </p>
+      ) : null}
+    </div>
+  )
+}
diff --git a/frontend/scripts/test-workflow-lifecycle-strip.mjs b/frontend/scripts/test-workflow-lifecycle-strip.mjs
new file mode 100644
index 0000000..55d8382
--- /dev/null
+++ b/frontend/scripts/test-workflow-lifecycle-strip.mjs
@@ -0,0 +1,74 @@
+/**
+ * Focused regression test for WorkflowLifecycleStrip's state -> view logic.
+ * No test runner is configured in this project (no vitest/jest), so this
+ * follows the existing plain-script pattern (see generate-m3-theme.mjs):
+ * a standalone Node script run directly, using Node's built-in TypeScript
+ * support and the built-in assert module.
+ * Usage: node scripts/test-workflow-lifecycle-strip.mjs
+ */
+import assert from 'node:assert/strict'
+
+import { deriveWorkflowLifecycleView } from '../components/studio/workflow-lifecycle-strip.logic.ts'
+
+let passed = 0
+
+function test(name, fn) {
+  fn()
+  passed += 1
+  console.log(`ok - ${name}`)
+}
+
+test('draft: activation stage carries the not-integrated note', () => {
+  const view = deriveWorkflowLifecycleView('draft')
+  assert.equal(view.primaryStatusLabel, '草稿')
+  const activate = view.stages.find((s) => s.key === 'activate')
+  assert.equal(activate.status, 'unavailable')
+  assert.equal(activate.note, '待后端接入')
+})
+
+test('validating -> validated -> publishing labels are distinct', () => {
+  assert.equal(deriveWorkflowLifecycleView('validating').primaryStatusLabel, '验证中')
+  assert.equal(deriveWorkflowLifecycleView('validated').primaryStatusLabel, '已验证')
+  assert.equal(deriveWorkflowLifecycleView('publishing').primaryStatusLabel, '发布中')
+})
+
+test('published: publish stage is done but activation is still unavailable, never active', () => {
+  const view = deriveWorkflowLifecycleView('published')
+  assert.equal(view.primaryStatusLabel, '已发布')
+
+  const publish = view.stages.find((s) => s.key === 'publish')
+  assert.equal(publish.status, 'done')
+
+  const activate = view.stages.find((s) => s.key === 'activate')
+  assert.equal(activate.status, 'unavailable')
+  assert.notEqual(activate.status, 'active')
+  assert.notEqual(activate.status, 'done')
+  assert.equal(activate.note, '待后端接入')
+})
+
+test('blocked: exactly one blocker message surfaces and validate stage reports error', () => {
+  const view = deriveWorkflowLifecycleView('blocked', '节点端口类型不匹配')
+  assert.equal(view.primaryStatusLabel, '已阻塞')
+  assert.equal(view.blockerText, '节点端口类型不匹配')
+
+  const validate = view.stages.find((s) => s.key === 'validate')
+  assert.equal(validate.status, 'error')
+})
+
+test('blocker text is only surfaced when state is actually blocked', () => {
+  const view = deriveWorkflowLifecycleView('draft', 'stale blocker text should be ignored')
+  assert.equal(view.blockerText, undefined)
+})
+
+test('every state exposes exactly four stages in a stable order', () => {
+  const states = ['draft', 'validating', 'validated', 'publishing', 'published', 'blocked']
+  for (const state of states) {
+    const view = deriveWorkflowLifecycleView(state)
+    assert.deepEqual(
+      view.stages.map((s) => s.key),
+      ['draft', 'validate', 'publish', 'activate'],
+    )
+  }
+})
+
+console.log(`\n${passed} passed`)
diff --git a/tests/integration/test_workflow_authoring_api.py b/tests/integration/test_workflow_authoring_api.py
new file mode 100644
index 0000000..d4b9c31
--- /dev/null
+++ b/tests/integration/test_workflow_authoring_api.py
@@ -0,0 +1,191 @@
+"""HTTP-seam tests for the Workspace -> Project -> WorkflowDraft -> WorkflowVersion
+persistence closed loop."""
+
+from __future__ import annotations
+
+from copy import deepcopy
+
+import pytest
+
+from tests.fixtures.workflow_conformance import workflow_conformance_project
+
+
+async def _create_workspace(client, *, slug: str = "macro-desk") -> dict:
+    response = await client.post(
+        "/api/v1/workspaces",
+        json={"name": "Macro Desk", "slug": slug, "description": "Macro workflows"},
+    )
+    assert response.status_code == 201, response.text
+    return response.json()["data"]
+
+
+async def _create_project(client, workspace_id: str, *, slug: str = "jin10-watch") -> dict:
+    response = await client.post(
+        f"/api/v1/workspaces/{workspace_id}/projects",
+        json={"name": "JIN10 Watch", "slug": slug},
+    )
+    assert response.status_code == 201, response.text
+    return response.json()["data"]
+
+
+async def _create_draft(client, project_id: str, *, snapshot: dict | None = None) -> dict:
+    response = await client.post(
+        f"/api/v1/projects/{project_id}/drafts",
+        json={"name": "Draft v1", "snapshot": snapshot or workflow_conformance_project()},
+    )
+    assert response.status_code == 201, response.text
+    return response.json()["data"]
+
+
+@pytest.mark.asyncio
+async def test_workspace_project_draft_validation_publish_closed_loop(client):
+    workspace = await _create_workspace(client)
+    settings_response = await client.get(f"/api/v1/workspaces/{workspace['id']}/settings")
+    assert settings_response.status_code == 200
+    settings = settings_response.json()["data"]
+    assert settings["timezone"] == "Asia/Shanghai"
+    assert settings["deterministic_simulation"] is True
+
+    project = await _create_project(client, workspace["id"])
+    draft = await _create_draft(client, project["id"])
+    assert draft["revision"] == 1
+
+    validation_response = await client.post(
+        f"/api/v1/drafts/{draft['id']}/validation-runs", json={}
+    )
+    assert validation_response.status_code == 201, validation_response.text
+    validation_run = validation_response.json()["data"]
+    assert validation_run["compile_valid"] is True
+    assert validation_run["status"] == "passed"
+
+    get_run_response = await client.get(
+        f"/api/v1/drafts/{draft['id']}/validation-runs/{validation_run['id']}"
+    )
+    assert get_run_response.status_code == 200
+    assert get_run_response.json()["data"]["id"] == validation_run["id"]
+
+    publish_response = await client.post(
+        f"/api/v1/drafts/{draft['id']}/publish",
+        json={"validation_run_id": validation_run["id"], "expected_revision": 1},
+    )
+    assert publish_response.status_code == 200, publish_response.text
+    version = publish_response.json()["data"]
+    assert version["version_number"] == 1
+    assert version["project_id"] == project["id"]
+
+    versions_response = await client.get(f"/api/v1/projects/{project['id']}/versions")
+    assert versions_response.status_code == 200
+    versions = versions_response.json()["data"]
+    assert len(versions) == 1
+    assert versions[0]["id"] == version["id"]
+
+    version_response = await client.get(f"/api/v1/versions/{version['id']}")
+    assert version_response.status_code == 200
+    assert version_response.json()["data"]["id"] == version["id"]
+
+
+@pytest.mark.asyncio
+async def test_draft_update_rejects_stale_revision(client):
+    workspace = await _create_workspace(client, slug="macro-desk-2")
+    project = await _create_project(client, workspace["id"], slug="jin10-watch-2")
+    draft = await _create_draft(client, project["id"])
+
+    snapshot = workflow_conformance_project()
+    ok_response = await client.put(
+        f"/api/v1/drafts/{draft['id']}",
+        json={"snapshot": snapshot, "expected_revision": 1},
+    )
+    assert ok_response.status_code == 200, ok_response.text
+    assert ok_response.json()["data"]["revision"] == 2
+
+    stale_response = await client.put(
+        f"/api/v1/drafts/{draft['id']}",
+        json={"snapshot": snapshot, "expected_revision": 1},
+    )
+    assert stale_response.status_code == 409
+
+
+@pytest.mark.asyncio
+async def test_publish_rejects_already_consumed_validation_run(client):
+    workspace = await _create_workspace(client, slug="macro-desk-3")
+    project = await _create_project(client, workspace["id"], slug="jin10-watch-3")
+    draft = await _create_draft(client, project["id"])
+
+    validation_run = (
+        await client.post(f"/api/v1/drafts/{draft['id']}/validation-runs", json={})
+    ).json()["data"]
+    assert validation_run["status"] == "passed"
+
+    first_publish = await client.post(
+        f"/api/v1/drafts/{draft['id']}/publish",
+        json={"validation_run_id": validation_run["id"], "expected_revision": 1},
+    )
+    assert first_publish.status_code == 200, first_publish.text
+
+    second_publish = await client.post(
+        f"/api/v1/drafts/{draft['id']}/publish",
+        json={"validation_run_id": validation_run["id"], "expected_revision": 1},
+    )
+    assert second_publish.status_code == 409
+
+
+@pytest.mark.asyncio
+async def test_publish_rejects_validation_run_stale_after_draft_update(client):
+    workspace = await _create_workspace(client, slug="macro-desk-4")
+    project = await _create_project(client, workspace["id"], slug="jin10-watch-4")
+    draft = await _create_draft(client, project["id"])
+
+    validation_run = (
+        await client.post(f"/api/v1/drafts/{draft['id']}/validation-runs", json={})
+    ).json()["data"]
+
+    await client.put(
+        f"/api/v1/drafts/{draft['id']}",
+        json={"snapshot": workflow_conformance_project(), "expected_revision": 1},
+    )
+
+    publish_response = await client.post(
+        f"/api/v1/drafts/{draft['id']}/publish",
+        json={"validation_run_id": validation_run["id"], "expected_revision": 2},
+    )
+    assert publish_response.status_code == 409
+
+
+@pytest.mark.asyncio
+async def test_workspace_settings_update_persists(client):
+    workspace = await _create_workspace(client, slug="macro-desk-5")
+
+    update_response = await client.put(
+        f"/api/v1/workspaces/{workspace['id']}/settings",
+        json={"timezone": "UTC", "max_items_per_run": 50},
+    )
+    assert update_response.status_code == 200, update_response.text
+    updated = update_response.json()["data"]
+    assert updated["timezone"] == "UTC"
+    assert updated["max_items_per_run"] == 50
+
+    get_response = await client.get(f"/api/v1/workspaces/{workspace['id']}/settings")
+    assert get_response.json()["data"]["timezone"] == "UTC"
+
+
+@pytest.mark.asyncio
+async def test_validation_run_fails_on_invalid_snapshot(client):
+    workspace = await _create_workspace(client, slug="macro-desk-6")
+    project = await _create_project(client, workspace["id"], slug="jin10-watch-6")
+    broken_snapshot = deepcopy(workflow_conformance_project())
+    broken_snapshot["edges"].append(
+        {"id": "e-dangling", "source": "does-not-exist", "target": "agent-normalize"}
+    )
+    draft = await _create_draft(client, project["id"], snapshot=broken_snapshot)
+
+    response = await client.post(f"/api/v1/drafts/{draft['id']}/validation-runs", json={})
+    assert response.status_code == 201, response.text
+    validation_run = response.json()["data"]
+    assert validation_run["status"] == "failed"
+    assert validation_run["failure_reason"] == "compile_failed"
+
+    publish_response = await client.post(
+        f"/api/v1/drafts/{draft['id']}/publish",
+        json={"validation_run_id": validation_run["id"], "expected_revision": 1},
+    )
+    assert publish_response.status_code == 409
diff --git a/tests/integration/test_workflow_compile_api.py b/tests/integration/test_workflow_compile_api.py
index c14f305..97ce28c 100644
--- a/tests/integration/test_workflow_compile_api.py
+++ b/tests/integration/test_workflow_compile_api.py
@@ -81,6 +81,29 @@ def _valid_workflow_project() -> dict:
     }
 
 
+def _nested_package_project(levels: int) -> dict:
+    def node_at(index: int) -> dict:
+        if index > levels:
+            return {
+                "id": f"leaf{index}",
+                "kind": "agent",
+                "capability": "normalize",
+                "params": {"language": "zh-CN"},
+            }
+        return {
+            "id": f"pkg{index}",
+            "kind": "agent",
+            "capability": "normalize",
+            "params": {"language": "zh-CN"},
+            "internals": {"nodes": [node_at(index + 1)], "edges": []},
+        }
+
+    project = _valid_workflow_project()
+    project["nodes"] = [node_at(1)]
+    project["edges"] = []
+    return project
+
+
 def _opencli_workflow_project() -> dict:
     project = _valid_workflow_project()
     project["nodes"][0] = {
@@ -856,6 +879,57 @@ async def test_compile_rejects_invalid_package_parameter_binding(client):
     ]
 
 
+@pytest.mark.asyncio
+async def test_compile_expands_two_level_nested_package(client):
+    project = _nested_package_project(2)
+
+    response = await client.post("/api/v1/workflows/compile", json={"project": project})
+
+    assert response.status_code == 200
+    data = response.json()["data"]
+    assert data["valid"] is True
+    runtime = data["plan"]["runtime"]
+    assert runtime["node_ids"] == ["pkg1", "pkg1::pkg2", "pkg1::pkg2::leaf3"]
+    assert runtime["edges"] == []
+    nodes_by_id = {node["id"]: node for node in runtime["nodes"]}
+    assert nodes_by_id["pkg1::pkg2"]["depends_on"] == ["pkg1"]
+    assert nodes_by_id["pkg1::pkg2::leaf3"]["depends_on"] == ["pkg1::pkg2"]
+    assert nodes_by_id["pkg1::pkg2::leaf3"]["runtime"]["package_parent_id"] == "pkg1::pkg2"
+
+
+@pytest.mark.asyncio
+async def test_compile_accepts_package_nesting_at_max_depth(client):
+    project = _nested_package_project(16)
+
+    response = await client.post("/api/v1/workflows/compile", json={"project": project})
+
+    assert response.status_code == 200
+    data = response.json()["data"]
+    assert data["valid"] is True
+    node_ids = data["plan"]["runtime"]["node_ids"]
+    assert len(node_ids) == 17
+    assert node_ids[-1] == "::".join([f"pkg{i}" for i in range(1, 17)] + ["leaf17"])
+
+
+@pytest.mark.asyncio
+async def test_compile_rejects_package_nesting_beyond_max_depth(client):
+    project = _nested_package_project(17)
+
+    response = await client.post("/api/v1/workflows/compile", json={"project": project})
+
+    assert response.status_code == 200
+    data = response.json()["data"]
+    assert data["valid"] is False
+    errors = [
+        error for error in data["errors"] if error["code"] == "package_nesting_limit_exceeded"
+    ]
+    assert len(errors) == 1
+    expected_node_id = "::".join([f"pkg{i}" for i in range(1, 18)])
+    assert errors[0]["node_id"] == expected_node_id
+    assert errors[0]["path"][:2] == ["nodes", "pkg1"]
+    assert errors[0]["path"][-1] == "internals"
+
+
 @pytest.mark.asyncio
 async def test_compile_records_existing_node_library_origin(client):
     project = _valid_workflow_project()
diff --git a/tests/integration/test_workflow_external_import_api.py b/tests/integration/test_workflow_external_import_api.py
new file mode 100644
index 0000000..4748885
--- /dev/null
+++ b/tests/integration/test_workflow_external_import_api.py
@@ -0,0 +1,134 @@
+"""HTTP-seam tests for Dify/n8n external workflow import."""
+
+from __future__ import annotations
+
+import pytest
+
+from backend.schemas.workflow import WorkflowProjectNode
+from backend.workflow.node_registry import resolve_node_origin
+
+
+def _base_project() -> dict:
+    return {
+        "id": "wf-import-target",
+        "name": "Import target",
+        "profile": "intelligence",
+        "version": 1,
+        "nodes": [
+            {
+                "id": "inbox-existing",
+                "kind": "inbox",
+                "capability": "store",
+                "params": {"queue": "macro-watch"},
+            }
+        ],
+        "edges": [],
+    }
+
+
+def _find_node(nodes: list[dict], node_id: str) -> dict:
+    return next(node for node in nodes if node["id"] == node_id)
+
+
+@pytest.mark.asyncio
+async def test_import_n8n_graph_tags_provenance_and_flattens_connections(client):
+    graph = {
+        "nodes": [
+            {"id": "1", "name": "Webhook", "type": "n8n-nodes-base.webhook"},
+            {"id": "2", "name": "Merge", "type": "n8n-nodes-base.merge"},
+        ],
+        "connections": {
+            "Webhook": {"main": [[{"node": "Merge", "type": "main", "index": 0}]]},
+        },
+    }
+
+    response = await client.post(
+        "/api/v1/workflows/import/external-runtime",
+        json={"project": _base_project(), "runtime": "n8n", "graph": graph, "name": "n8n export"},
+    )
+
+    assert response.status_code == 200, response.text
+    data = response.json()["data"]
+    assert data["valid"] is True, data["errors"]
+
+    nodes = data["project"]["nodes"]
+    webhook_node = _find_node(nodes, "webhook")
+    merge_node = _find_node(nodes, "merge")
+
+    assert webhook_node["ui"]["catalogId"] == "external.tool.capability"
+    assert webhook_node["ui"]["n8n"] == {
+        "source": "n8n",
+        "nodeId": "Webhook",
+        "nodeType": "n8n-nodes-base.webhook",
+    }
+    assert merge_node["ui"]["catalogId"] == "intelligence.flow.merge"
+    assert merge_node["ui"]["n8n"] == {
+        "source": "n8n",
+        "nodeId": "Merge",
+        "nodeType": "n8n-nodes-base.merge",
+    }
+
+    edges = data["project"]["edges"]
+    imported_edge = next(e for e in edges if e["source"] == "webhook" and e["target"] == "merge")
+    assert imported_edge["ui"]["externalWorkflow"]["runtime"] == "n8n"
+
+    # catalogId resolution takes precedence over the ui.n8n provenance tag (see
+    # resolve_node_origin), so importer output always classifies as node_library —
+    # the n8n tag is provenance metadata only, not a classification switch.
+    assert resolve_node_origin(WorkflowProjectNode.model_validate(webhook_node)).kind == (
+        "node_library"
+    )
+    assert resolve_node_origin(WorkflowProjectNode.model_validate(merge_node)).kind == (
+        "node_library"
+    )
+
+
+@pytest.mark.asyncio
+async def test_import_dify_graph_tags_provenance_without_n8n_marker(client):
+    graph = {
+        "nodes": [
+            {"id": "node-1", "type": "llm", "name": "LLM call"},
+            {"id": "node-2", "type": "transform", "name": "Normalize output"},
+        ],
+        "edges": [{"source": "node-1", "target": "node-2"}],
+    }
+
+    response = await client.post(
+        "/api/v1/workflows/import/external-runtime",
+        json={"project": _base_project(), "runtime": "dify", "graph": graph, "name": "dify export"},
+    )
+
+    assert response.status_code == 200, response.text
+    data = response.json()["data"]
+    assert data["valid"] is True, data["errors"]
+
+    nodes = data["project"]["nodes"]
+    llm_node = _find_node(nodes, "node-1")
+    normalize_node = _find_node(nodes, "node-2")
+
+    assert llm_node["ui"]["externalWorkflow"]["runtime"] == "dify"
+    assert "n8n" not in llm_node["ui"]
+    assert normalize_node["ui"]["catalogId"] == "intelligence.processing.normalize"
+    assert "n8n" not in normalize_node["ui"]
+
+    edges = data["project"]["edges"]
+    imported_edge = next(e for e in edges if e["source"] == "node-1" and e["target"] == "node-2")
+    assert imported_edge["ui"]["externalWorkflow"]["runtime"] == "dify"
+
+
+def test_resolve_node_origin_classifies_unmapped_n8n_tagged_node():
+    node = WorkflowProjectNode(
+        id="raw-n8n-node",
+        kind="action",
+        capability="store",
+        ui={"n8n": {"source": "n8n", "nodeId": "Raw", "nodeType": "n8n-nodes-base.rawUnmapped"}},
+    )
+
+    origin = resolve_node_origin(node)
+
+    assert origin.kind == "n8n"
+    assert origin.n8n == {
+        "source": "n8n",
+        "nodeId": "Raw",
+        "nodeType": "n8n-nodes-base.rawUnmapped",
+    }