Native PipelineWorkspace and event broker in periscope/src.

PCB and placement jobs import job_workspace instead of PinScope
pipeline.py. Analysis run_pipeline re-exports the same broker singleton.

periscope/dependency/backend/services/pipeline.py

@@ -95,58 +95,20 @@ def _git_commit() -> str:
 
 
 # ---------------------------------------------------------------------------
-# SSE Event Broker
+# SSE Event Broker + workspace live in periscope/src job_workspace.py.
-# ---------------------------------------------------------------------------
+# Re-export so analysis run_pipeline and pipeline_worker.set_broker stay
+# on one singleton. PinScope pipeline.py is not empty-deleted.
+from backend.services import job_workspace as _job_ws
 
-class EventBroker:
+EventBroker = _job_ws.EventBroker
-    """In-memory pub/sub for SSE events, keyed by project_id.
+PipelineWorkspace = _job_ws.PipelineWorkspace
-
+broker = _job_ws.broker
-    Buffers all events per project so late subscribers (e.g. after a page
-    navigation) receive the full history before seeing live events.
-    """
-
-    def __init__(self):
-        self._queues: dict[str, list[asyncio.Queue]] = {}
-        self._history: dict[str, list[dict]] = {}
-
-    def subscribe(self, project_id: str) -> asyncio.Queue:
-        q: asyncio.Queue = asyncio.Queue()
-        # Replay buffered events so the subscriber catches up
-        for msg in self._history.get(project_id, []):
-            q.put_nowait(msg)
-        self._queues.setdefault(project_id, []).append(q)
-        return q
-
-    def unsubscribe(self, project_id: str, q: asyncio.Queue) -> None:
-        qs = self._queues.get(project_id, [])
-        if q in qs:
-            qs.remove(q)
-        if not qs:
-            self._queues.pop(project_id, None)
-
-    def clear_history(self, project_id: str) -> None:
-        self._history.pop(project_id, None)
-
-    def publish(self, project_id: str, event: str, data: dict) -> None:
-        msg = {"event": event, "data": data}
-        self._history.setdefault(project_id, []).append(msg)
-        for q in self._queues.get(project_id, []):
-            q.put_nowait(msg)
 
 
-broker: EventBroker = EventBroker()
+def set_broker(b) -> None:
-
+    """Swap the worker event broker (GCS in prod). Updates native + this module."""
-
+    _job_ws.set_broker(b)
-def set_broker(b: EventBroker) -> None:
+    globals()["broker"] = b
-    """Replace the module-level broker.
-
-    Called by :mod:`backend.pipeline_worker` at startup to swap in the
-    GCS-backed broker so events written from the worker are visible to
-    the API's SSE handler. Must be called *before* :func:`run_pipeline`
-    or :func:`run_regen_pipeline`.
-    """
-    global broker
-    broker = b
 
 
 # Per-process cancel-flag cache: re-reading the project meta from GCS on
@@ -189,160 +151,6 @@ def _cancel_gate_check(ctx: PipelineContext) -> None:
         raise CancelRequested(f"cancel requested for {ctx.project_id}")
 
 
-# ---------------------------------------------------------------------------
-# Pipeline Workspace
-# ---------------------------------------------------------------------------
-
-class PipelineWorkspace:
-    """Downloads project files from storage to a temp dir for pipeline execution.
-
-    The periscopex core library operates on local paths. This context manager
-    downloads inputs at enter, provides local paths, and uploads results at exit.
-    """
-
-    def __init__(
-        self,
-        storage: StorageBackend,
-        user_id: str,
-        project_id: str,
-    ) -> None:
-        self.storage = storage
-        self.user_id = user_id
-        self.project_id = project_id
-        self.prefix = proj_svc.project_prefix(user_id, project_id)
-        self._tmpdir: tempfile.TemporaryDirectory | None = None
-        self.local_dir: Path = Path()
-
-    async def __aenter__(self) -> PipelineWorkspace:
-        self._tmpdir = tempfile.TemporaryDirectory()
-        self.local_dir = Path(self._tmpdir.name)
-
-        # Create subdirectories
-        (self.local_dir / "uploads" / "datasheets").mkdir(parents=True)
-        (self.local_dir / "extracted").mkdir(parents=True)
-        (self.local_dir / "patterns").mkdir(parents=True)
-        (self.local_dir / "models").mkdir(parents=True)
-        (self.local_dir / "taxonomy").mkdir(parents=True)
-
-        # Download project files from storage
-        all_keys = self.storage.list_recursive(self.prefix)
-        for key in all_keys:
-            # key is like users/{uid}/projects/{pid}/uploads/bom.csv
-            # We want the relative part after the project prefix
-            rel = key[len(self.prefix) + 1:]  # strip prefix + trailing /
-            local_path = self.local_dir / rel
-            self.storage.download_to_local(key, local_path)
-
-        # Download taxonomy files from storage
-        taxonomy_keys = self.storage.list_prefix("taxonomy/")
-        for key in taxonomy_keys:
-            if key.endswith(".json"):
-                filename = key.rsplit("/", 1)[-1]
-                self.storage.download_to_local(key, self.local_dir / "taxonomy" / filename)
-
-        # Seed from repo taxonomy if storage had no taxonomy files yet
-        local_tax = self.local_dir / "taxonomy"
-        if not any(local_tax.glob("*.json")):
-            repo_tax = settings.taxonomy_dir
-            if repo_tax.is_dir():
-                for f in repo_tax.glob("*.json"):
-                    shutil.copy2(f, local_tax / f.name)
-
-        return self
-
-    async def __aexit__(self, exc_type, exc_val, exc_tb):
-        if exc_type is None:
-            # Upload outputs back to storage
-            self._upload_dir("extracted")
-            self._upload_dir("patterns")
-            self._upload_dir("models")
-            self._upload_file("design_graph.json")
-            self._upload_file("layout_graph.json")
-            self._upload_file("impedance_nets.json")
-            self._upload_file("functional_groups.json")
-            self._upload_file("placement_plan.json")
-            self._upload_file("bom_summary.json")
-            self._upload_file("derating.json")
-            self._upload_file("report.json")
-            self._upload_file("pcb_report.json")
-            self._upload_file("pcb_inventory.json")
-            self._upload_file("periscope-findings.json")
-            self._upload_file("review_fingerprints.json")
-            self._upload_file("api_logs.jsonl")
-
-            # Merge taxonomy: read current from storage, add any new entries
-            # from this run, write back.  This avoids clobbering subtypes
-            # that a concurrent pipeline added while we were running.
-            tax_dir = self.local_dir / "taxonomy"
-            if tax_dir.is_dir():
-                for f in tax_dir.iterdir():
-                    if f.is_file() and f.suffix == ".json":
-                        local_data = json.loads(f.read_text())
-                        local_subtypes = local_data.get("subtypes", {})
-                        storage_key = f"taxonomy/{f.name}"
-
-                        if self.storage.exists(storage_key):
-                            current = self.storage.read_json(storage_key)
-                            merged = current.get("subtypes", {})
-                            for key, entry in local_subtypes.items():
-                                if key not in merged:
-                                    merged[key] = entry
-                                else:
-                                    # Backfill fields the local run generated
-                                    # (e.g. specs_schema) that the
-                                    # storage copy is missing.
-                                    for field, value in entry.items():
-                                        if field not in merged[key]:
-                                            merged[key][field] = value
-                            current["subtypes"] = merged
-                            self.storage.write_json(storage_key, current)
-                        else:
-                            self.storage.write_json(storage_key, local_data)
-
-        if self._tmpdir:
-            self._tmpdir.cleanup()
-
-    def _upload_dir(self, subdir: str) -> None:
-        """Upload all files in a subdirectory back to storage."""
-        local = self.local_dir / subdir
-        if not local.is_dir():
-            return
-        for f in local.rglob("*"):
-            if f.is_file():
-                rel = f.relative_to(self.local_dir)
-                key = f"{self.prefix}/{rel}"
-                self.storage.upload_from_local(f, key)
-
-    def _upload_file(self, name: str) -> None:
-        """Upload a single file back to storage if it exists."""
-        local = self.local_dir / name
-        if local.is_file():
-            self.storage.upload_from_local(local, f"{self.prefix}/{name}")
-
-    def local_path(self, rel: str) -> Path:
-        """Get a local path within the workspace."""
-        return self.local_dir / rel
-
-    def netlist_local_path(self) -> Path:
-        """Local path of whichever netlist file was synced (``.asc`` or ``.edn``).
-
-        Pipeline workspace mirrors the entire project prefix, so whichever
-        format the user uploaded lands locally with its original extension.
-        Falls back to ``uploads/netlist.asc`` if neither exists — downstream
-        code will raise a clearer error when it tries to read the missing
-        file than a ``None`` return would.
-        """
-        for ext in ("asc", "edn", "xml", "kicad_net", "kicad_sch"):
-            p = self.local_dir / "uploads" / f"netlist.{ext}"
-            if p.exists():
-                return p
-        return self.local_dir / "uploads" / "netlist.asc"
-
-    @property
-    def taxonomy_dir(self) -> Path:
-        return self.local_dir / "taxonomy"
-
-
 # ---------------------------------------------------------------------------
 # Pipeline context — shared state threaded through all stage functions
 # ---------------------------------------------------------------------------

periscope/dependency/frontend/content/changelog.md

@@ -2,6 +2,13 @@
 
 What's new in Periscope.
 
+## 2.41.0 — 2026-09-20 — Native job workspace for PCB and placement
+
+`pcb_pipeline` and `placement_pipeline` no longer import PinScope `pipeline.py`. Workspace download/upload and the in-memory event broker live in `periscope/src` (`job_workspace.py`). Analysis `run_pipeline` still lives in `dependency/` and re-exports the same singleton so `pipeline_worker.set_broker` keeps PCB SSE on the GCS event log. Parsers/graph unchanged. `pipeline.py` not deleted.
+
+- [New] `backend.services.job_workspace`: `PipelineWorkspace`, `EventBroker`, `set_broker`.
+- [Changed] PCB/placement import `job_workspace`; `pipeline.py` re-exports the native types.
+
 ## 2.40.0 — 2026-09-20 — Fase C3: native datasheet extraction
 
 Pipeline pintable/pattern/specs/auto-resolve run from `periscope/src` (`datasheet_extract.py`): DeepSeek + local SKILL.md, no Anthropic Console skill ids. Inherited `extraction.py` stays in `dependency/`.

periscope/src/backend/services/job_workspace.py

@@ -0,0 +1,192 @@
+"""Native project workspace + in-memory job event broker.
+
+PCB and placement jobs need a local mirror of storage and an SSE broker
+without importing PinScope ``pipeline.py`` (parsers, review, extraction).
+``pipeline_worker.set_broker`` must update this module's ``broker`` so PCB
+events still go to the GCS event log.
+
+Inherited ``backend.services.pipeline`` re-exports these names after the
+slice is proven. Do not empty-delete ``pipeline.py``.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import shutil
+import tempfile
+from pathlib import Path
+
+from backend.config import settings
+from backend.services import projects as proj_svc
+from backend.services.storage import StorageBackend
+
+
+class EventBroker:
+    """In-memory pub/sub for SSE events, keyed by project_id.
+
+    Buffers all events per project so late subscribers (e.g. after a page
+    navigation) receive the full history before seeing live events.
+    """
+
+    def __init__(self):
+        self._queues: dict[str, list[asyncio.Queue]] = {}
+        self._history: dict[str, list[dict]] = {}
+
+    def subscribe(self, project_id: str) -> asyncio.Queue:
+        q: asyncio.Queue = asyncio.Queue()
+        for msg in self._history.get(project_id, []):
+            q.put_nowait(msg)
+        self._queues.setdefault(project_id, []).append(q)
+        return q
+
+    def unsubscribe(self, project_id: str, q: asyncio.Queue) -> None:
+        qs = self._queues.get(project_id, [])
+        if q in qs:
+            qs.remove(q)
+        if not qs:
+            self._queues.pop(project_id, None)
+
+    def clear_history(self, project_id: str) -> None:
+        self._history.pop(project_id, None)
+
+    def publish(self, project_id: str, event: str, data: dict) -> None:
+        msg = {"event": event, "data": data}
+        self._history.setdefault(project_id, []).append(msg)
+        for q in self._queues.get(project_id, []):
+            q.put_nowait(msg)
+
+
+broker: EventBroker = EventBroker()
+
+
+def set_broker(b: EventBroker) -> None:
+    """Replace the module-level broker (GCS-backed in the worker)."""
+    global broker
+    broker = b
+
+
+class PipelineWorkspace:
+    """Downloads project files from storage to a temp dir for pipeline execution.
+
+    The periscopex core library operates on local paths. This context manager
+    downloads inputs at enter, provides local paths, and uploads results at exit.
+    """
+
+    def __init__(
+        self,
+        storage: StorageBackend,
+        user_id: str,
+        project_id: str,
+    ) -> None:
+        self.storage = storage
+        self.user_id = user_id
+        self.project_id = project_id
+        self.prefix = proj_svc.project_prefix(user_id, project_id)
+        self._tmpdir: tempfile.TemporaryDirectory | None = None
+        self.local_dir: Path = Path()
+
+    async def __aenter__(self) -> PipelineWorkspace:
+        self._tmpdir = tempfile.TemporaryDirectory()
+        self.local_dir = Path(self._tmpdir.name)
+
+        (self.local_dir / "uploads" / "datasheets").mkdir(parents=True)
+        (self.local_dir / "extracted").mkdir(parents=True)
+        (self.local_dir / "patterns").mkdir(parents=True)
+        (self.local_dir / "models").mkdir(parents=True)
+        (self.local_dir / "taxonomy").mkdir(parents=True)
+
+        all_keys = self.storage.list_recursive(self.prefix)
+        for key in all_keys:
+            rel = key[len(self.prefix) + 1:]
+            local_path = self.local_dir / rel
+            self.storage.download_to_local(key, local_path)
+
+        taxonomy_keys = self.storage.list_prefix("taxonomy/")
+        for key in taxonomy_keys:
+            if key.endswith(".json"):
+                filename = key.rsplit("/", 1)[-1]
+                self.storage.download_to_local(key, self.local_dir / "taxonomy" / filename)
+
+        local_tax = self.local_dir / "taxonomy"
+        if not any(local_tax.glob("*.json")):
+            repo_tax = settings.taxonomy_dir
+            if repo_tax.is_dir():
+                for f in repo_tax.glob("*.json"):
+                    shutil.copy2(f, local_tax / f.name)
+
+        return self
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb):
+        if exc_type is None:
+            self._upload_dir("extracted")
+            self._upload_dir("patterns")
+            self._upload_dir("models")
+            self._upload_file("design_graph.json")
+            self._upload_file("layout_graph.json")
+            self._upload_file("impedance_nets.json")
+            self._upload_file("functional_groups.json")
+            self._upload_file("placement_plan.json")
+            self._upload_file("bom_summary.json")
+            self._upload_file("derating.json")
+            self._upload_file("report.json")
+            self._upload_file("pcb_report.json")
+            self._upload_file("pcb_inventory.json")
+            self._upload_file("periscope-findings.json")
+            self._upload_file("review_fingerprints.json")
+            self._upload_file("api_logs.jsonl")
+
+            tax_dir = self.local_dir / "taxonomy"
+            if tax_dir.is_dir():
+                for f in tax_dir.iterdir():
+                    if f.is_file() and f.suffix == ".json":
+                        local_data = json.loads(f.read_text())
+                        local_subtypes = local_data.get("subtypes", {})
+                        storage_key = f"taxonomy/{f.name}"
+
+                        if self.storage.exists(storage_key):
+                            current = self.storage.read_json(storage_key)
+                            merged = current.get("subtypes", {})
+                            for key, entry in local_subtypes.items():
+                                if key not in merged:
+                                    merged[key] = entry
+                                else:
+                                    for field, value in entry.items():
+                                        if field not in merged[key]:
+                                            merged[key][field] = value
+                            current["subtypes"] = merged
+                            self.storage.write_json(storage_key, current)
+                        else:
+                            self.storage.write_json(storage_key, local_data)
+
+        if self._tmpdir:
+            self._tmpdir.cleanup()
+
+    def _upload_dir(self, subdir: str) -> None:
+        local = self.local_dir / subdir
+        if not local.is_dir():
+            return
+        for f in local.rglob("*"):
+            if f.is_file():
+                rel = f.relative_to(self.local_dir)
+                key = f"{self.prefix}/{rel}"
+                self.storage.upload_from_local(f, key)
+
+    def _upload_file(self, name: str) -> None:
+        local = self.local_dir / name
+        if local.is_file():
+            self.storage.upload_from_local(local, f"{self.prefix}/{name}")
+
+    def local_path(self, rel: str) -> Path:
+        return self.local_dir / rel
+
+    def netlist_local_path(self) -> Path:
+        for ext in ("asc", "edn", "xml", "kicad_net", "kicad_sch"):
+            p = self.local_dir / "uploads" / f"netlist.{ext}"
+            if p.exists():
+                return p
+        return self.local_dir / "uploads" / "netlist.asc"
+
+    @property
+    def taxonomy_dir(self) -> Path:
+        return self.local_dir / "taxonomy"

periscope/src/backend/services/pcb_pipeline.py

@@ -29,7 +29,7 @@ from backend.periscopex.models import (
 from backend.periscopex.pcb_checks import assign_pcb_finding_ids, run_pcb_checks
 from backend.periscopex.pcb_inventory import build_pcb_inventory
 from backend.services import projects as proj_svc
-from backend.services.pipeline import PipelineWorkspace, broker
+from backend.services.job_workspace import PipelineWorkspace, broker
 from backend.services.storage import StorageBackend
 
 logger = logging.getLogger(__name__)

periscope/src/backend/services/placement_pipeline.py

@@ -16,7 +16,7 @@ from backend.periscopex.graph import build_graph
 from backend.periscopex.models import ComponentConstraints, DesignGraph, LayoutGraph
 from backend.periscopex.placement_pack import build_placement_pack
 from backend.services import projects as proj_svc
-from backend.services.pipeline import PipelineWorkspace, broker
+from backend.services.job_workspace import PipelineWorkspace, broker
 from backend.services.storage import StorageBackend
 
 logger = logging.getLogger(__name__)

periscope/src/docs/development/PINSCOPE_INDEPENDENCE_PLAN.md

@@ -1,6 +1,6 @@
 # Piano — indipendenza architettonica e di licenza da PinScope
 
-**Stato:** split **2.38.0**. C2 review **2.39.0/2.39.1** live. C3 extraction **2.40.0**. `validate.py` / `extraction.py` restano in `dependency/` (non chiamati dal live path). Fork non staccato.  
+**Stato:** split **2.38.0**. C2 review **2.39.x**. C3 extraction **2.40.0**. C4 PCB off `validate.py`. **2.41.0** native `job_workspace` — PCB/placement non importano `pipeline.py`. Parsers/graph still dependency. Fork non staccato.  
 **Gate Michele:** sostituire/smettere di chiamare un modulo `dependency/` solo dopo pytest + deploy smoke. Se la verifica fallisce, resta il path ereditato.  
 **Sequenza:** split → sostituzione incrementale (C2 loop → C3 extraction → C4 PCB off `validate.py`). **Mai** empty-delete. Parsers/graph (C5/E) e auto-place fuori scope. AGPL resta.
 
@@ -60,6 +60,7 @@ Trattare come **una dipendenza in-tree**, non come prodotto Periscope:
 | Auth self-host | `periscope/src/backend/services/local_jwt.py`, `routers/auth.py` | REPLACEMENT |
 | LLM DeepSeek | `periscope/src/backend/services/llm/deepseek_provider.py`, `local_skill.py`, `pdf_ingest.py` | NEW |
 | Review loop C2 | `review_session.py`, `review_parse.py`, `review_tools.py`, `review_context.py`, `constraints_lookup.py` | REPLACEMENT 2.39.0; PinScope files kept |
+| Job workspace | `periscope/src/backend/services/job_workspace.py` | REPLACEMENT 2.41.0; PCB/placement off `pipeline.py` |
 | Deploy | `scripts/update-periscope.sh`, `docker-compose.yml`, `periscope/src/backend/Dockerfile` | NEW (nomi `pinscope_*` ancora WEAK) |
 | Plugin KiCad | `periscope/src/plugins/kicad/` | NEW |
 | Check deterministici fork | `dnp_check`, `sequencing_check`, `layout_rules`, … under `periscope/src` | NEW ma **INDIRECT**: usano `models` / graph |
@@ -226,6 +227,7 @@ Qui sì si riscrive l’engine ereditato. Ordine interno:
 | C1 | Adapter: `validate.py` emette solo `Finding` grezzi → `complete_finding` | Già parziale; chiudere i campi doppi |
 | C2 | **REWRITE** loop per-IC native in `periscope/src` (DeepSeek, tools su `DesignGraph`) | **Shipped 2.39.0** live `review_ic_async` → `review_session`; PinScope files **kept**, not called from live loop |
 | C3 | Extraction: `local_skill.py` + schemi JSON (KEEP schema se identici; REWRITE orchestrazione Anthropic) | **Shipped 2.40.0** live `datasheet_extract`; PinScope `extraction.py` kept |
+| C3b | Native `PipelineWorkspace` / event broker | **Shipped 2.41.0** `job_workspace.py`; `pipeline.py` kept as analysis orchestrator |
 | C4 | Spegnere import da `validate.py` nel PCB (`_parse_review` → parser finding nativo) | **Shipped 2.39.1** `pcb_validation` → `review_parse` / `review_session` |
 | C5 | Test golden `simple_project` + Emmaforo: parity FACT/REQUIREMENT, non parity prose | |
 

tests/test_native_job_workspace.py

@@ -0,0 +1,113 @@
+"""Prove native workspace/broker before PCB/placement stop importing pipeline.py."""
+
+from __future__ import annotations
+
+import ast
+import inspect
+from pathlib import Path
+
+import pytest
+
+from backend.services.job_workspace import (
+    EventBroker,
+    PipelineWorkspace,
+    set_broker,
+)
+from backend.services.job_workspace import broker as native_broker
+from backend.services.pipeline import EventBroker as InheritedBroker
+from backend.services.pipeline import PipelineWorkspace as InheritedWorkspace
+from backend.services.storage import LocalStorageBackend
+
+
+def test_native_job_workspace_does_not_import_pipeline():
+    import backend.services.job_workspace as jw
+
+    tree = ast.parse(Path(jw.__file__).read_text())
+    imported = [
+        node.module
+        for node in ast.walk(tree)
+        if isinstance(node, ast.ImportFrom) and node.module
+    ]
+    assert "backend.services.pipeline" not in imported
+    assert "backend.periscopex.graph" not in imported
+    assert "backend.periscopex.parsers" not in imported
+
+
+def test_event_broker_publish_subscribe_and_replay():
+    b = EventBroker()
+    b.publish("p1", "pcb_step_update", {"stage": "parse_pcb"})
+    q = b.subscribe("p1")
+    msg = q.get_nowait()
+    assert msg == {"event": "pcb_step_update", "data": {"stage": "parse_pcb"}}
+    b.publish("p1", "pcb_complete", {"findings": 1})
+    msg2 = q.get_nowait()
+    assert msg2["event"] == "pcb_complete"
+    b.unsubscribe("p1", q)
+    b.clear_history("p1")
+    q2 = b.subscribe("p1")
+    assert q2.empty()
+
+
+def test_set_broker_swaps_module_singleton():
+    original = native_broker
+    replacement = EventBroker()
+    set_broker(replacement)
+    import backend.services.job_workspace as jw
+    assert jw.broker is replacement
+    set_broker(original)
+    assert jw.broker is original
+
+
+@pytest.mark.asyncio
+async def test_workspace_mirrors_storage_and_netlist_like_inherited(tmp_path):
+    storage = LocalStorageBackend(tmp_path)
+    prefix = "users/local/projects/ws1"
+    storage.write_json(f"{prefix}/project.json", {"id": "ws1"})
+    bom = tmp_path / "bom.csv"
+    bom.write_text("Reference,MPN\nU1,PART\n")
+    storage.upload_from_local(bom, f"{prefix}/uploads/bom.csv")
+    net = tmp_path / "netlist.edn"
+    net.write_text("(edif dummy)")
+    storage.upload_from_local(net, f"{prefix}/uploads/netlist.edn")
+
+    native_net = None
+    inherited_net = None
+    async with PipelineWorkspace(storage, "local", "ws1") as nws:
+        native_net = nws.netlist_local_path()
+        assert nws.local_path("uploads/bom.csv").is_file()
+        assert native_net.name == "netlist.edn"
+        nws.local_path("pcb_report.json").write_text("{}")
+        nws._upload_file("pcb_report.json")
+    async with InheritedWorkspace(storage, "local", "ws1") as iws:
+        inherited_net = iws.netlist_local_path()
+        assert inherited_net.name == native_net.name
+    assert storage.exists(f"{prefix}/pcb_report.json")
+
+
+def test_pcb_and_placement_do_not_import_pipeline_py():
+    import backend.services.pcb_pipeline as pcb
+    import backend.services.placement_pipeline as place
+
+    for mod in (pcb, place):
+        tree = ast.parse(Path(mod.__file__).read_text())
+        imported = [
+            node.module
+            for node in ast.walk(tree)
+            if isinstance(node, ast.ImportFrom) and node.module
+        ]
+        assert "backend.services.pipeline" not in imported
+        assert "backend.services.job_workspace" in imported
+
+
+def test_pipeline_reexports_native_workspace():
+    from backend.services import job_workspace as jw
+    from backend.services import pipeline as pipe
+    assert pipe.PipelineWorkspace is jw.PipelineWorkspace
+    assert pipe.EventBroker is jw.EventBroker
+    assert inspect.signature(EventBroker.publish) == inspect.signature(
+        InheritedBroker.publish
+    )
+    native_methods = {"local_path", "netlist_local_path", "_upload_file", "_upload_dir"}
+    for name in native_methods:
+        assert hasattr(InheritedWorkspace, name)
+        assert hasattr(PipelineWorkspace, name)