plan: add resource system guidance

fix: change reset and submit_experiment* from manual confirm to normal node
Fix: update sirna_station to support new flexible resultTable
2026-05-23 20:51:23 +00:00 · 2026-05-20 19:17:48 +08:00 · 2026-05-15 15:25:54 +08:00 · 2026-05-15 14:01:34 +08:00 · 2026-05-12 19:35:39 +08:00 · 2026-05-12 19:35:39 +08:00
10 changed files with 6537 additions and 17 deletions
--- a/plan/2026-05-20_11_resource_material_sync_guidance_draft.md
+++ b/plan/2026-05-20_11_resource_material_sync_guidance_draft.md
@@ -0,0 +1,914 @@
 # Draft: Resource And Material Sync Guidance For Sirna And Similar Bioyond Systems
 Status: draft for discussion, not an implementation mandate.
 This note triangulates across five source categories with different authority:
 1. `docs/developer_guide/examples/workstation_architecture.md`: desired shape
   and vocabulary. It is a design target, not proof that the current code has
   every behavior.
 2. `BioyondWorkstation`, `BioyondResourceSynchronizer`, and shared graphio code:
   the current compatibility anchor. New guidance should mostly preserve this
   lifecycle and extend it deliberately.
 3. Existing non-Sirna Bioyond stations: practical examples of how the shared
   base is used, including shortcuts that should not become policy.
 4. Sirna implementation, plans, findings, and guide notes: stress-test evidence.
   They expose real missing cases, but the current Sirna code is not the
   architecture authority because parts were added without fully aligning to the
   shared base.
 5. UniLabOS resource framework behavior around PLR resources,
   `ResourceTreeSet`, serialize/deserialize, and `update_resource(resources=...)`.
 The short recommendation is:
 Keep the shared Bioyond workstation lifecycle as the center of gravity. Evolve
 the shared synchronizer with small project hooks for classification, ID-based
 resolution, and non-slot material handling, rather than letting Sirna become a
 parallel synchronization model. For Sirna, those hooks should resolve placement
 by Bioyond IDs, distinguish physical slot labware from reagent liquid contents,
 preserve Bioyond IDs in `unilabos_extra`, mutate the local PLR deck, and publish
 the full deck through `update_resource(resources=[deck])`.
 Do not treat every Bioyond stock row as a deck resource.
 ## Evidence Weighting
 This task is not a Sirna implementation retrospective. It is a best-practice
 alignment pass across the architecture target, the shared base class, observed
 station behavior, and Sirna's newly exposed edge cases.
 Use the sources this way:
 - Architecture doc: ask "what shape should this system eventually have?"
 - Shared Bioyond base: ask "what behavior must remain compatible today?"
 - Other Bioyond stations: ask "what patterns are already working in practice?"
 - Sirna current code and notes: ask "what did the base model fail to handle, and
  which Sirna fixes conflict with the shared lifecycle?"
 - Live API/schema evidence: ask "what is true for this deployment's material,
  warehouse, and coordinate data?"
 The Sirna AGENT_GUIDE is useful for finding caveats and prior investigations,
 but it should not be cited as an independent source of truth when source code,
 framework behavior, live API evidence, or architecture docs can answer the same
 question.
 ## Mental Model
 There are three resource worlds. Confusing them is the main source of bugs.
 | World | Owner | Purpose | Recommended truth |
 | --- | --- | --- | --- |
 | Bioyond/LIMS | Bioyond APIs through `BioyondV1RPC` | External stock, material IDs, warehouse IDs, location IDs, inbound/outbound side effects | External material truth |
 | PLR deck | Workstation driver | Runtime workstation material layout, warehouse occupancy, liquid contents | Local mutation surface |
 | UniLabOS resource tree | `ResourceTreeSet` / ROS node / host resource APIs | Canonical UniLabOS/cloud representation | Framework/cloud truth |
 The architecture guide describes `Deck` as the local material system and
 `ResourceSynchronizer` as the optional external-system bridge
 (`docs/developer_guide/examples/workstation_architecture.md:221`). The broader
 framework serializes PLR objects into `ResourceTreeSet`, whose resource dicts
 carry `id`, `uuid`, `parent_uuid`, `type`, `class`, `pose`, `config`, `data`,
 and `extra` (`unilabos/resources/resource_tracker.py:107`).
 Therefore the correct path is:
 ```text
 Bioyond rows
  -> normalized material records
  -> preserve Bioyond materialTypeMode: Sample / Consumables / Reagent
  -> resolve material/location/warehouse IDs
  -> choose UniLabOS handling for that mode
  -> mutate PLR deck
  -> ResourceTreeSet.from_plr_resources([deck])
  -> ROS update_resource(resources=[deck])
  -> host/cloud resource-tree update
 ```
 Avoid direct cloud JSON patches and avoid treating Bioyond records as already
 being UniLabOS resource nodes.
 ## Target Architecture
 The ideal architecture in `workstation_architecture.md` is still the right
 direction:
 1. `WorkstationBase` owns the local `deck`, workflow state, and hardware
   interface.
 2. `ResourceSynchronizer` owns external material synchronization.
 3. `BioyondResourceSynchronizer` owns the Bioyond-specific use of
   `BioyondV1RPC`.
 4. `ROS2WorkstationNode.update_resource(resources)` owns the UniLabOS/cloud
   resource-tree update.
 5. Optional HTTP report handlers may trigger local deck mutation, external sync,
   and cloud publication.
 The documented startup sequence is:
 1. Create workstation.
 2. Initialize PLR deck and warehouses.
 3. Create Bioyond RPC hardware interface.
 4. Create resource synchronizer.
 5. `sync_from_external()`.
 6. Initialize ROS node and children.
 7. `post_init(ros_node)`.
 8. Upload `resources=[deck]`.
 See `docs/developer_guide/examples/workstation_architecture.md:308`,
 `docs/developer_guide/examples/workstation_architecture.md:497`, and
 `docs/developer_guide/examples/workstation_architecture.md:737`.
 Important caveat: that document is the hope. It is still valuable because it
 names the desired responsibilities, but current Bioyond stations implement a
 more limited, best-effort side-effect sync. When the doc and current code
 diverge, use the doc to choose direction and the shared base code to choose the
 next compatible step.
 ## Current Practical Behavior
 The shared `BioyondWorkstation` implementation is the current behavioral
 baseline. It does this today:
 1. Requires a deck.
 2. Reconstructs `deck.warehouses` from deck children/config when needed.
 3. Creates `BioyondV1RPC`.
 4. Installs `BioyondResourceSynchronizer`.
 5. Immediately calls `sync_from_external()`.
 6. Later, in `post_init`, publishes the whole deck with
   `ROS2DeviceNode.run_async_func(self._ros_node.update_resource, True,
   resources=[self.deck])`.
 Relevant code:
 - `BioyondResourceSynchronizer` is defined in
  `unilabos/devices/workstation/bioyond_studio/station.py:117`.
 - `sync_from_external()` fetches stock `typeMode` 0, 1, and 2, then passes all
  rows to `resource_bioyond_to_plr(...)`
  (`unilabos/devices/workstation/bioyond_studio/station.py:147`).
 - `BioyondWorkstation.__init__` installs the synchronizer and syncs immediately
  (`unilabos/devices/workstation/bioyond_studio/station.py:856`).
 - `post_init()` uploads the deck (`unilabos/devices/workstation/bioyond_studio/station.py:893`).
 - `resource_tree_add()` performs Bioyond create/inbound side effects
  (`unilabos/devices/workstation/bioyond_studio/station.py:958`).
 This practical behavior works for simple physical-resource stock import, but it
 does not provide full continuous two-way sync, conflict resolution, or reliable
 stale-state cleanup. In particular:
 - Re-fetching stock does not clearly clear stale local deck state first.
 - Local-to-external update no-ops unless `unilabos_extra["update_resource_site"]`
  is present.
 - `process_material_change_report()` is mostly TODO-level in the base station.
 - Some station-specific code bypasses the shared synchronizer with direct LIMS
  calls.
 - Some existing stations contain shortcuts that should not be copied, such as
  duplicate initialization, stale globals, or hardcoded warehouse/axis cases.
 Treat current Bioyond behavior as operationally useful and compatibility
 important, not as proof that the ideal architecture is already achieved. The
 goal is to tighten this base path, not replace it with a Sirna-only path.
 ## Recommended Shared Pipeline
 For Sirna and similar Bioyond systems, the base synchronizer should evolve from
 the current shared path into one shared pipeline with small project hooks:
 ```text
 sync_from_external()
  fetch stock rows from Bioyond
  update RPC material cache
  normalize rows into a common internal shape
  preserve materialTypeMode: Sample / Consumables / Reagent
  resolve warehouse/location by Bioyond IDs when available
  choose handling for the row's mode
  apply Sample / Consumables rows as mapped slot labware by default
  apply Reagent rows as physical reagent labware or liquid content by evidence
  report unknown modes or unmapped handling loudly
  publish deck if deck changed and ROS node is available
 ```
 Suggested extension points:
 ```python
 class BioyondResourceSynchronizer(ResourceSynchronizer):
    def external_material_mode(self, row: dict) -> str:
        return row["materialTypeMode"]  # Sample | Consumables | Reagent
    def resolve_external_row(self, row: dict) -> dict:
        ...
    def apply_external_row(self, row: dict, mode: str, resolved: dict) -> None:
        ...
 ```
 The base class should continue to own:
 - stock fetches for `typeMode` 0/1/2;
 - material-cache refresh;
 - common normalization;
 - mode validation for `Sample` / `Consumables` / `Reagent`;
 - publication orchestration;
 - error aggregation;
 - stale-state policy once agreed.
 Project code should own only the parts that are truly deployment-specific:
 - material type/mode to PLR class mapping;
 - project-local handling inside `Sample` / `Consumables` / `Reagent`;
 - project-local warehouse ID/name mapping;
 - project-local coordinate conventions;
 - special row handling such as Sirna reagent-as-liquid.
 This keeps existing Bioyond stations close to the same lifecycle while still
 absorbing the Sirna lessons that the earlier base did not model. The default
 hook behavior can remain "Sample/Consumables/Reagent rows become mapped slot
 labware" for simpler stations; Sirna should override Reagent handling where
 evidence requires liquid-content behavior.
 ## Sirna Findings To Feed Back Into Shared Design
 Treat Sirna as a stress test for the shared base, not as a replacement design.
 It raises real questions that earlier stations did not need to answer:
 - within the `Reagent` mode, some external rows may be liquid contents rather
  than slot-occupying reagent labware;
 - placement should be ID-first where Bioyond supplies material/location IDs;
 - warehouse and axis metadata must survive serialize/deserialize;
  project-specific mode handling is available.
 For Sirna-like deployments, the old implicit rule "stock row equals PLR
 resource" is too coarse. For simpler deployments, it can remain the default
 mode-handling behavior.
 ### IDs Win
 Placement identity should prefer:
 ```text
 materialId
 locationId
 materialTypeId
 warehouseId / whid
 ```
 Display/debug fields are not identity:
 ```text
 materialName
 materialCode
 locationCode
 locationShowName
 warehouseName / whName
 ```
 `locationCode` such as `1-1` is not globally unique. It can exist in multiple
 warehouses. Code-only resolution may be kept as a diagnostic fallback, but it
 must raise on ambiguity.
 The current Sirna ID-first resolver is useful source evidence for this direction
 (`unilabos/devices/workstation/bioyond_studio/sirna_station/sirna_station.py:3659`).
 The Sirna mega plan captures the same concern, but the implementation and live
 Bioyond IDs are the stronger evidence.
 ### Bioyond Modes And UniLabOS Handling Are Different
 Bioyond has three primary material modes in this context:
 ```text
 Sample
 Consumables
 Reagent
 ```
 Those modes should be preserved as the external taxonomy. UniLabOS still needs a
 handling decision inside the mode: should the row create/place a physical PLR
 resource, or update contents on an already-placed parent resource?
 Physical slot resources include things like:
 - tip racks;
 - plates;
 - cell culture plates;
 - empty trough/bottle labware;
 - other objects that occupy a warehouse slot.
 Reagents are contents of a parent labware when the row describes a liquid rather
 than a physical holder. For Sirna, Bioyond `stock_material(typeMode=2)` returns
 `materialTypeMode="Reagent"` rows; those rows still need evidence-based handling
 as either physical reagent labware or reagent liquid content. The finding in
 `temp_benyao/sirna/_findings/2026-05-07_reagents_vs_resources.md:6` records the
 bug: the generic path can fall back to `RegularContainer`, fail registry lookup,
 and drop reagent rows such as `试剂槽裂解液/Betame`.
 Recommended behavior:
 1. Validate `materialTypeMode` as `Sample`, `Consumables`, or `Reagent`.
 2. For `Sample` and `Consumables`, default to mapped slot labware: instantiate
   the mapped PLR class and place it into the resolved warehouse slot.
 3. For `Reagent`, decide whether the row represents physical reagent labware or
   reagent liquid content from material type evidence, row shape, and live data.
 4. For physical reagent labware, place the mapped PLR resource in the resolved
   slot.
 5. For reagent liquid content, find the parent trough/bottle and attach liquid
   metadata to its tracker.
 6. Preserve Bioyond IDs in `parent.unilabos_extra["reagent_bioyond_ids"]`.
 7. Make liquid attachment idempotent by Bioyond material ID.
 8. If the parent labware is missing or the mode/handling cannot be resolved,
   defer or log with IDs; do not guess.
 The existing Sirna helper `_attach_liquid_to_parent()` already follows the
 idempotent metadata direction
 (`unilabos/devices/workstation/bioyond_studio/sirna_station/sirna_station.py:4074`).
 ### The `0003` Question Must Stay Evidence-Based
 Do not blindly map `materialTypeCode 0003` to `BioyondSirna_ReagentTrough`.
 If live/schema evidence says a `0003` row is physical trough labware, map it to
 a PLR resource class. If evidence says it is liquid content, attach it to the
 parent trough. If evidence is contradictory, mark it unsupported and log the
 Bioyond IDs.
 Treat this as an open design decision until source/schema/live evidence settles
 the row shape. The Sirna plan records the question, but it should not decide the
 mapping by itself.
 ### Sirna Warehouse And Axis Rules Are Project-Local
 Sirna warehouse layout and axis conventions must be verified from Sirna source,
 Sirna schema, current deck behavior, and live/read-only Sirna APIs when
 available. Do not import Peptide, reaction, dispensing, or cell station layout
 truth.
 For Sirna specifically, the current integrated deck display should be treated as
 the good baseline. The prior col-row slot-key fix is already present in the
 deck/graphio path, and any remaining x/y or y-reverse correction should be
 handled through shared warehouse metadata and shared graphio/display mapping
 rather than by reshaping the Sirna station display ad hoc.
 ### Display Geometry: Evidence Before Axis Values
 Peptide resource sync should not be treated as a validated model, but its
 UniLabOS display work is useful and should influence this guidance.
 Confirmed Peptide behavior:
 - Peptide live warehouse evidence found `自动化堆栈` with 170 locations where
  `code="10-17"` appears at `x=17, y=10`
  (`../Uni-Lab-OS-Peptide/temp_benyao/peptide/_findings/2026-05-13_1404_peptide_col_row_deck.md:6`).
 - Peptide's current display convention is Peptide-specific evidence:
  `bioyond_axis="xy_col_row"` and `bioyond_key_axis="col_row"` in the current
  Peptide station implementation. Do not copy that pair into Sirna or any other
  project without live warehouse evidence for that target system. With this
  Peptide combination, graphio does not apply the legacy x/y swap
  (`../Uni-Lab-OS-Peptide/unilabos/resources/graphio.py:868`).
 - Peptide models the main automation stack as 17 visual rows by 10 visual
  columns while preserving labels such as `10-17`
  (`../Uni-Lab-OS-Peptide/unilabos/resources/bioyond/decks.py:308`).
 - Peptide warehouses and deck child positions use `frontend_y_flip=True` /
  `_frontend_y_flipped_coordinate(...)` so stored PLR coordinates compensate for
  the frontend y-axis inversion
  (`../Uni-Lab-OS-Peptide/unilabos/resources/bioyond/decks.py:299`;
  `temp_benyao/peptide/_findings/2026-05-13_1514_frontend_y_flip_layout.md:6`).
 - The older Peptide graph-layout note is stale for the Sirna discussion. The
  current Sirna integrated station/deck display is a good baseline; the reusable
  Peptide lesson is axis metadata and frontend y-reverse compatibility.
 - Peptide tests encode the intended behavior: `10-17` lands at row 17 /
  column 10, and after frontend y-flip the displayed site positions match the
  expected top-to-bottom layout
  (`../Uni-Lab-OS-Peptide/temp_benyao/peptide/tests/test_peptide_deck_layout.py:60`).
 Guidance for Sirna and similar systems:
 1. Treat `bioyond_axis` and `bioyond_key_axis` as two separate concepts:
   - `bioyond_axis` describes how Bioyond numeric `x/y` map to visual axes.
   - `bioyond_key_axis` describes how slot labels such as `10-17` are generated
     or preserved.
 2. Do not infer display orientation from label strings alone. Both `row_col`
   and `col_row` can preserve the same final label text while producing
   different visual layouts and graphio swap behavior.
 3. Use live/read-only `code/x/y` evidence for each project and each warehouse
   family before choosing axis metadata.
 4. Keep visual orientation fixes separate from material identity. IDs and slot
   labels determine registration; display dimensions and frontend y-flip only
   determine how the deck appears.
 5. Author intended display coordinates first, then convert stored y coordinates
   for the active frontend convention:
   - deck child stored y = `deck_height - display_y - child_height`;
   - warehouse site stored y = `warehouse_height - display_y - site_height`;
   - graph-level y values should be transformed only when the active frontend
     convention requires it.
 6. Preserve the current Sirna display as the baseline unless concrete frontend
   evidence shows a y-reverse problem. Do not change station/deck graph
   semantics just to match stale Peptide layout notes.
 7. Add layout tests that assert:
   - warehouse `num_items_x`, `num_items_y`, capacity, first key, and last key;
   - representative `code/x/y` examples land on the intended site key;
   - frontend y-flip produces expected displayed positions;
   - generated deck children do not overlap in displayed coordinates.
 8. When borrowing from Peptide, borrow the evidence pattern: live discovery,
   axis/key-axis metadata, y-flip tests, and display fixtures. Do not borrow a
   literal axis pair or Peptide resource-sync behavior as a validated path.
 Concrete live discovery workflow:
 1. Prefer the reusable read-only probe pattern before reading old findings:
 ```bash
 python3 temp_benyao/sirna/tests/probe_readonly_storage_inventory.py \
  --base-url <api_host> \
  --api-key <api_key> \
  --output temp_benyao/<project>/_logs/<timestamp>_readonly_storage_inventory_probe.json
 ```
 This probe checks swagger candidates, project storage/location endpoints,
 material type endpoints, `warehouse-info-by-mat-type-id`, and `stock-material`,
 then writes a redacted evidence file and a merged warehouse summary.
 2. For Sirna quick checks, `temp_benyao/sirna/tests/discover_sirna_warehouses.py`
   is the narrower historical helper. Treat it as a template unless it has been
   parameterized for the target config.
 3. Query `/api/storage/location/locations-by-type?type=0&typeMode=0&materialType=0`
   first for stack names, warehouse IDs, full slot lists, `code`, `x`, `y`, `z`,
   and display mode. This endpoint is the topology source when available.
 4. Cross-reference with `/api/lims/storage/material-types` and
   `/api/lims/storage/warehouse-info-by-mat-type-id` to learn material-type
   placement constraints. Use `stock-material` only for occupied-slot evidence;
   it cannot reveal empty topology.
 5. Infer `bioyond_key_axis` from how slot labels must be generated or preserved,
   and infer `bioyond_axis` from how raw Bioyond `x/y` must map to PLR holder
   indices. A label such as `10-17` alone is ambiguous; compare it to the same
   record's `x/y`, and use boundary examples from non-square stacks.
 6. Encode the discovered convention on the warehouse resource, not in a one-off
   station branch. The metadata must serialize/deserialize with the warehouse
   because graph load and cloud sync rebuild resources.
 7. Add or update layout tests before changing Sirna display behavior. For Sirna,
   test against the current good display first, then only change shared x/y or
   y-reverse mapping if the fixture demonstrates a real mismatch.
 ### Deserialize Must Be Idempotent
 This is confirmed framework behavior, not a possible risk. Resource publication
 builds `ResourceTreeSet` from live PLR resources by calling
 `resource.serialize()` and `resource.serialize_all_state()`
 (`unilabos/resources/resource_tracker.py:547`). Readback/reconstruction goes
 through `ResourceTreeSet.to_plr_resources()`, finds the PLR subclass, calls
 `sub_cls.deserialize(...)`, then restores PLR state, UUIDs, and `unilabos_extra`
 (`unilabos/resources/resource_tracker.py:637`). `ResourceTreeSet.dump()` also
 serializes resource nodes while excluding `children` from each individual node
 record (`unilabos/resources/resource_tracker.py:914`), so parent/child
 relationships and object state must survive the framework tree shape rather than
 only an in-memory PLR object graph.
 Sirna resource/deck classes therefore must survive:
 1. registry-time construction;
 2. `Resource.deserialize()` round trips;
 3. cloud-synced deck state with serialized children.
 If a synced deck already has serialized `children`, default setup must not
 create duplicate/stale child resources. Sirna noticed this problem, but the
 reason is framework-level: cloud/resource-tree sync reconstructs PLR objects
 from serialized state, so constructors and setup logic must be idempotent.
 Existing Sirna deck code already acknowledges the issue: `BIOYOND_Sirna_Deck`
 turns `setup=False` during deserialize when serialized `children` are present
 (`unilabos/resources/bioyond/decks.py:153`). Sirna material classes also use
 registry-visible `@resource(...)` decorators and tolerant constructors with
 `*args, **kwargs`, defaults, and `ordering` fallbacks
 (`unilabos/resources/bioyond/sirna_materials.py:14`). Treat those as required
 patterns for any new synced resource class.
 Guidance:
 - Do not add a deck/resource class until it round-trips through
  `Resource.serialize()` / `Resource.deserialize()` and
  `ResourceTreeSet.from_plr_resources(...).to_plr_resources()`.
 - Preserve `unilabos_uuid`, parentage, `unilabos_extra`, and liquid state across
  the round trip.
 - Make default setup conditional: if serialized children exist, do not recreate
  default warehouses or default child resources on top of them.
 - Itemized PLR subclasses must provide `ordered_items` or `ordering`; otherwise
  deserialization can fail or rebuild a different structure.
 ## Startup And Resync Recommendation
 The current Sirna code installs `SirnaResourceSynchronizer` after
 `super().post_init(ros_node)` (`sirna_station.py:363`). That is useful as a
 phase patch and as evidence that classification hooks are needed, but it should
 not become the long-term lifecycle. Base initialization may already have run
 generic stock sync and base post-init may already have published the deck before
 the Sirna synchronizer is installed.
 Recommended pathway:
 1. Keep the `BioyondWorkstation` lifecycle as the base path.
 2. Add a synchronizer factory or hook registration point before eager sync, for
   example `create_resource_synchronizer()`.
 3. Let Sirna provide classification/resolution hooks through that shared
   synchronizer shape.
 4. Make startup sync use the installed hook-aware synchronizer before the first
   deck publication.
 5. Make manual resync, reset resync, report-triggered resync, and future
   periodic sync all call `self.resource_synchronizer.sync_from_external()`.
 6. Do not instantiate fresh base synchronizers inside Sirna actions, because
   that bypasses the installed project logic.
 This directly addresses the overlap finding in
 `temp_benyao/sirna/_findings/2026-05-12_sirna_synchronizer_overlap.md:6`.
 ## Material Registration From Create-Order Results
 Create-order allocation registration should remain separate from stock sync, but
 it should use the same classification and apply logic.
 Recommended create-order pipeline:
 1. Normalize allocation records into:
 ```python
 {
    "materialId": "...",
    "materialCode": "...",
    "materialName": "...",
    "materialTypeId": "...",
    "materialTypeCode": "...",
    "materialTypeMode": "Sample|Consumables|Reagent",
    "materialTypeName": "...",
    "locationId": "...",
    "locationCode": "...",
    "locationShowName": "...",
 }
 ```
 2. Resolve warehouse and slot by:
 ```text
 material-info(materialId)
  -> warehouse-info-by-mat-type-id(materialTypeId) matched by locationId
  -> code-only diagnostic fallback only if unambiguous
 ```
 3. Classify as `slot_labware`, `liquid_content`, or `unsupported`.
 4. Apply mutation to the PLR deck.
 5. Publish the full deck once after the batch.
 The current Sirna `_register_materials_to_tree()` already documents this flow
 (`sirna_station.py:3659`) and should be the local reference implementation until
 the shared hook is designed.
 ## Cloud / UniLabOS Publication Rules
 Always mutate real tracked PLR resources first. Then publish through the
 framework.
 Correct call shape:
 ```python
 ROS2DeviceNode.run_async_func(
    self._ros_node.update_resource,
    True,
    **{"resources": [self.deck]},
 )
 ```
 The real method is:
 ```python
 async def update_resource(self, resources: List["ResourcePLR"])
 ```
 See `unilabos/ros/nodes/base_device_node.py:727`.
 Do not call:
 ```python
 update_resource(resource_name=..., resource_data=...)
 ```
 Do not manually serialize the deck for this path. `update_resource()` creates a
 `ResourceTreeSet`, sends it to `/c2s_update_resource_tree`, and applies returned
 UUID mappings. Missing root parent UUIDs are auto-mounted to the current device,
 so parentage should be preserved on PLR objects before publication.
 ## Metadata Contract
 Every Bioyond-originated slot resource should carry enough metadata for unload,
 audit, and later sync:
 ```python
 plr_resource.unilabos_extra = {
    "material_bioyond_id": mat["materialId"],
    "material_bioyond_code": mat["materialCode"],
    "material_bioyond_name": mat["materialName"],
    "material_bioyond_type_id": mat["materialTypeId"],
    "material_bioyond_type_code": mat["materialTypeCode"],
    "material_bioyond_type_mode": mat["materialTypeMode"],
    "location_bioyond_id": mat["locationId"],
    "location_code": resolved["location_code"],
    "warehouse_bioyond_id": resolved["warehouse_id"],
    "warehouse_bioyond_name": resolved["warehouse_name"],
    "location_resolution_source": resolved["source"],
 }
 ```
 Every Bioyond-originated liquid content should preserve equivalent metadata on
 the parent labware:
 ```python
 parent.unilabos_extra.setdefault("reagent_bioyond_ids", []).append({
    "material_bioyond_id": mat["materialId"],
    "material_bioyond_code": mat["materialCode"],
    "material_bioyond_name": mat["materialName"],
    "material_bioyond_type_id": mat["materialTypeId"],
    "material_bioyond_type_code": mat["materialTypeCode"],
    "location_bioyond_id": mat["locationId"],
    "quantity": mat.get("quantity"),
    "location_resolution_source": resolved["source"],
 })
 ```
 This metadata is not decorative. It is required for correct unload, audit,
 duplicate prevention, and future round-trip sync.
 ## Stale State And Conflict Policy
 This is confirmed current behavior, not just a theoretical risk.
 Current stock import places resources into empty warehouse slots and skips
 occupied slots. That prevents overwrites, but it can leave stale local resources
 after external moves/deletes.
 Evidence:
 - `BioyondResourceSynchronizer.sync_from_external()` fetches Bioyond stock and
  delegates to `resource_bioyond_to_plr(...)`, but it does not compute a
  before/after diff or remove local resources absent from the snapshot
  (`unilabos/devices/workstation/bioyond_studio/station.py:147`).
 - `resource_bioyond_to_plr(...)` only assigns when the target warehouse position
  is empty or a placeholder; when a real resource already occupies the slot, it
  logs "跳过放置" and leaves the existing object in place
  (`unilabos/resources/graphio.py:936`).
 - Bioyond outbound/removal logic exists in separate local-to-external hooks such
  as `resource_tree_remove(...)`, but that is not invoked by stock refresh for
  Bioyond rows that disappeared externally
  (`unilabos/devices/workstation/bioyond_studio/station.py:987`).
 Recommended direction:
 1. Define Bioyond as the source of truth for external stock snapshots unless a
   local operation is in progress.
 2. Before applying a stock snapshot, compute a diff by Bioyond material ID.
 3. Remove or mark local resources whose Bioyond IDs disappeared from the
   snapshot, subject to workflow safety checks.
 4. Move local resources whose Bioyond location ID changed.
 5. Attach/detach liquid contents idempotently by Bioyond material ID.
 6. Publish once after applying the batch.
 7. In ambiguous or active-operation cases, log and require manual confirmation.
 Until this policy exists, call the current implementation "refresh/import" rather
 than "authoritative synchronization."
 ## External-To-Local And Local-To-External Boundaries
 External-to-local:
 - Bioyond stock and create-order allocation rows mutate the local PLR deck.
 - Then UniLabOS publication derives resource trees from the PLR deck.
 Local-to-external:
 - `resource_tree_add` / update / remove may call Bioyond add, inbound, outbound,
  or move APIs.
 - These paths should require Bioyond IDs or explicit creation parameters.
 - Movement should use `unilabos_extra["update_resource_site"]` only as an
  explicit request, not as hidden ambient state.
 - After successful Bioyond side effects, refresh or update the PLR deck and
  publish the deck.
 Avoid station-level direct LIMS calls that bypass the synchronizer unless the
 action explicitly reconciles the deck afterward.
 ## Error Handling Rules
 Fail loudly on:
 - unknown material type code/name;
 - unresolved warehouse ID/name;
 - location ID not found in `warehouse-info-by-mat-type-id`;
 - code-only location ambiguity;
 - missing parent labware for liquid content;
 - invalid PLR resource class or `Resource.deserialize()` failure;
 - Bioyond RPC returning empty/fallback values where an ID is required.
 Do not silently create `RegularContainer` to hide mapping failures. The Sirna
 finding at `temp_benyao/sirna/_findings/2026-05-07_reagents_vs_resources.md:52`
 calls this out as a reusable trap.
 ## Tests And Validation
 Minimum offline tests:
 1. `update_resource` is called only with `resources=[deck]`.
 2. `ResourceTreeSet.from_plr_resources([deck])` preserves UUIDs, parentage,
   `data`, and `unilabos_extra`.
 3. Sirna allocation records resolve by `material-info` first.
 4. `warehouse-info-by-mat-type-id` resolves by `locationId`.
 5. Code-only fallback raises on ambiguous slots.
 6. `Sample` and `Consumables` rows become slot labware.
 7. Reagent content rows become parent liquid metadata.
 8. Re-running registration does not duplicate liquid entries.
 9. Missing parent labware is deferred/logged, not guessed.
 10. Unknown material type emits Bioyond IDs.
 11. Deck deserialize does not recreate duplicate default children.
 12. Warehouse coordinate mapping is tested per warehouse, not globally.
 13. A synced Sirna deck round-trips through PLR deserialize and
    `ResourceTreeSet` conversion without duplicate default children.
 14. Stock refresh with a missing Bioyond material ID proves current add/skip
    behavior first, then verifies the chosen diff/delete policy once implemented.
 15. Display geometry tests cover project-local live discovery fixtures,
    `bioyond_axis`, `bioyond_key_axis`, representative `code/x/y` mappings,
    frontend y-flip, and no-overlap deck layout.
 Focused Sirna command:
 ```bash
 pytest temp_benyao/sirna/tests/test_sirna_resource_system.py -q
 ```
 Shared conversion command, when changing graphio or Bioyond converters:
 ```bash
 pytest tests/resources/test_converter_bioyond.py -q
 ```
 Live/read-only validation should capture fixtures for:
 - `stock_material(typeMode=0/1/2, includeDetail=true)`;
 - `material-info(materialId)`;
 - `warehouse-info-by-mat-type-id(materialTypeId)`;
 - create-order allocation records;
 - frontend/cloud resource-tree readback after publication.
 ## Discrepancies To Discuss
 ### 1. Architecture Is Direction; Base Code Is The Compatibility Anchor
 The architecture guide presents `ResourceSynchronizer` as bidirectional. Current
 Bioyond code is mostly eager external import plus selected add/remove/update
 side effects. It has no complete continuous conflict-resolution loop.
 Sirna current code should be read in this context: it exposes missing
 capabilities, but parts of it conflict with the existing base lifecycle because
 it was layered on after the base sync behavior already existed.
 Recommendation: keep the bidirectional architecture as the target, preserve
 `BioyondWorkstation` as the implementation anchor, and introduce shared hooks
 for the missing Sirna-class problems. Describe the current implementation as
 best-effort refresh plus Bioyond side effects until a diff/conflict policy
 exists.
 ### 2. The Doc Centers `Deck`; The Framework Centers `ResourceTreeSet`
 The architecture guide uses PLR `Deck` as the material system. UniLabOS resource
 tracking uses `ResourceTreeSet` as the canonical serialized representation.
 Recommendation: phrase the model as "Deck is the local PLR mutation surface;
 ResourceTreeSet is the UniLabOS/cloud representation derived from it."
 ### 3. Base Bioyond Sync Treats Every Stock Row As PLR Resource
 `BioyondResourceSynchronizer.sync_from_external()` currently feeds typeMode
 0/1/2 rows into `resource_bioyond_to_plr(...)`.
 Sirna evidence shows this is wrong for reagent/content-like rows.
 Recommendation: add classification hooks before conversion. Rows classified as
 liquid content must not go through standalone PLR resource conversion.
 ### 4. Sirna Fix Exists As A Fork, Not A Shared Hook
 Current `SirnaResourceSynchronizer` captures the important reagent-as-liquid
 question, but it duplicates or bypasses shared sync mechanics. Startup can run
 base sync before the Sirna synchronizer is installed, and manual/reset paths
 have used fresh base synchronizers.
 Recommendation: keep the discovered behavior where evidence supports it, but
 refactor the shape. A factory/hook in the shared synchronizer is preferable to a
 long-lived parallel sync class.
 ### 5. Sirna Create-Order Registration Is Stronger Than Stock Sync
 Create-order registration now has ID-first resolution and a clearer classifier.
 That is valuable evidence, not automatically the canonical sync design. The
 stock sync path still resolves external liquid rows by warehouse name/code and
 stores weaker metadata.
 Recommendation: extract the stronger resolver/classifier ideas into shared
 hooks, then converge stock sync and create-order registration on the same
 functions.
 ### 6. Graphio Has Shared Fragility
 `resource_bioyond_to_plr()` contains hardcoded warehouse/coordinate cases and a
 fallback to `RegularContainer`. Hardening it affects multiple Bioyond projects.
 Recommendation: after Sirna is green, discuss shared graphio hardening in a
 separate change: no silent `RegularContainer`, explicit unknown-type diagnostics,
 and project-specific axis metadata instead of hardcoded names.
 ### 7. Serialize/Deserialize Is A Gating Contract
 This is not optional documentation neatness. Any new resource/deck class or sync
 metadata must pass the framework serialize/deserialize path before it can be
 trusted in cloud-synced operation.
 Recommendation: make round-trip tests part of acceptance for resource-system
 changes, especially for Sirna deck setup, warehouse metadata, `reagent_bioyond_ids`,
 liquid tracker state, and itemized labware ordering.
 ### 8. Existing Stock Sync Is Add/Skip-Biased, Not Delete-Aware
 Current `sync_from_external()` imports and places observed Bioyond resources,
 but it does not delete local resources missing from the latest stock snapshot.
 When a target slot is occupied by a real resource, graphio skips placement rather
 than replacing it.
 Recommendation: describe the current behavior as "stock refresh/import" until a
 Bioyond-ID diff policy is implemented. Deletion/removal must be a deliberate
 phase, with workflow-safety checks and tests, not implied by the current stock
 sync name.
 ### 9. Peptide Display Is Useful; Peptide Sync Is Not Yet Authority
 Peptide's latest display work is useful evidence for Peptide's own warehouse
 axis/key-axis metadata and for the shared need to account for frontend y-axis
 inversion. It does not define Sirna's axis values. Its resource sync remains
 untested/problematic and should not be treated as the model for Sirna sync
 behavior.
 Recommendation: fold the Peptide display lesson into shared guidance as a
 discovery-and-test pattern, not as a literal axis pair. Keep Sirna display
 behavior grounded in Sirna live evidence and the current good deck baseline; keep
 resource synchronization recommendations grounded in the Sirna ID-first
 registration path and the shared Bioyond synchronizer refactor.
 ### 10. Some Existing Stations Should Be Examples Only, Not Templates
 Non-Sirna stations show useful patterns, but also shortcuts:
 - direct LIMS calls bypassing the synchronizer;
 - duplicate `super().__init__()` in reaction station;
 - stale/undefined `WAREHOUSE_MAPPING` references;
 - RPC methods collapsing failures into empty values;
 - no robust stale deck cleanup on stock refresh.
 Recommendation: borrow the shared deck publication and Bioyond ID metadata
 patterns, not the incidental shortcuts.
 ## Recommended Discussion Path
 For the next implementation discussion, decide these in order:
 1. What is the smallest shared-base extension that lets project hooks exist
   before eager sync without breaking existing Bioyond stations?
 2. Should `BioyondWorkstation` gain a synchronizer factory, a hook registry, or
   a delayed eager-sync point?
 3. Is `materialTypeCode 0003` in current Sirna live data physical labware,
   liquid content, or both depending on row shape?
 4. Where should Sirna warehouse Bioyond IDs live: station config, deck metadata,
   or both?
 5. Should stock refresh clear/diff deck state now, or remain append/skip until
   after create-order registration is stable?
 6. Should `update_resource` remain async/non-blocking for material registration,
   or should selected user-facing actions fail if publication fails?
 7. Which parts of the Sirna reagent-as-liquid rule are project-local, and which
   should become shared Bioyond behavior after Peptide/cell validation?
 8. Should graphio fallback hardening land before or after Sirna stock sync is
   fully validated?
 9. Should any remaining Sirna x/y or y-reverse issue be fixed in shared
   graphio/display mapping after a live fixture proves it, or should the current
   Sirna deck remain untouched?
 ## Recommended Pathway
 The pragmatic path is:
 1. Preserve and test the current `BioyondWorkstation` lifecycle as the shared
   compatibility baseline.
 2. Treat Sirna create-order registration as a local proof of the missing
   classification/ID-resolution behavior, not as the architecture shape to copy.
 3. Validate live/read-only Sirna fixtures for `0003`, warehouse IDs, and stock
   reagent rows.
 4. Refactor the shared Bioyond synchronizer to expose classification,
   resolution, and non-slot row hooks with default behavior matching existing
   stations.
 5. Make Sirna install hooks before the first external sync and first deck
   publication through the shared lifecycle.
 6. Route startup, manual resync, reset resync, and report-triggered resync
   through the installed synchronizer.
 7. Converge stock sync and create-order registration on one
   resolver/classifier/apply implementation.
 8. Add stale-state diffing only after the basic ID-first path is stable.
 9. Harden graphio fallback as a shared follow-up.
 This gives Sirna the necessary behavior without locking in a second Bioyond sync
 system that future projects will have to debug around.
--- a/unilabos/devices/workstation/bioyond_studio/debug_call_log.py
+++ b/unilabos/devices/workstation/bioyond_studio/debug_call_log.py
@@ -0,0 +1,459 @@
 """Per-action raw call/response log for Bioyond stations.
 When a debug session is active, ``wrap_rpc_http`` replaces a ``BioyondV1RPC``
 instance's ``post`` / ``get`` methods with closures that perform the HTTP
 transport themselves, capture the request/response details, and append a record
 to the active session before returning exactly what ``BaseRequest`` would have
 returned. Outside of an active session the wrapped method delegates to the
 original (unwrapped) implementation, leaving non-debug behavior intact.
 The session writes a Markdown file under ``out_dir`` mirroring the format of
 ``temp_benyao/peptide/_logs/2026-04-30_160316_day3_samplefile_only_raw_calls.md``
 minus the "Raw Payload Argument" section.
 This module has no dependency on ``BioyondV1RPC`` itself; the only contract is
 that the wrapped instance descends from ``BaseRequest`` (i.e. has a logger
 returned by ``self.get_logger()``).
 """
 from __future__ import annotations
 import contextvars
 import copy
 import inspect
 import json
 import re
 from contextlib import contextmanager
 from dataclasses import dataclass, field
 from datetime import datetime
 from pathlib import Path
 from typing import Any, Iterator, List, Optional
 import requests
 __all__ = [
    "CallRecord",
    "CallLogContext",
    "session",
    "wrap_rpc_http",
    "active_session",
 ]
 _DEFAULT_TIMEOUT_GET = 30
 _DEFAULT_TIMEOUT_POST = 120
@dataclass
 class CallRecord:
    """One captured HTTP call inside a debug session."""
    index: int
    method: str
    url: str
    path: str
    source: str
    transport: str
    http_status: Optional[int]
    request_body: Any
    response_body: Any
    error: Optional[str] = None
@dataclass
 class CallLogContext:
    """State for a single ``session()`` block.
    A session lazily creates its file on the first appended record. Actions
    that abort before any RPC produce no file.
    """
    action: str
    out_dir: Path
    started_at: datetime
    calls: List[CallRecord] = field(default_factory=list)
    file_path: Optional[Path] = None
    def append(self, record: CallRecord) -> None:
        record.index = len(self.calls) + 1
        self.calls.append(record)
        self._write_file()
    # -- file I/O -------------------------------------------------------------
    def _resolve_file_path(self) -> Path:
        if self.file_path is not None:
            return self.file_path
        timestamp = self.started_at.strftime("%Y-%m-%d_%H%M%S")
        slug = _slugify_action(self.action)
        candidate = self.out_dir / f"{timestamp}_{slug}_raw_calls.md"
        suffix = 2
        while candidate.exists():
            candidate = (
                self.out_dir
                / f"{timestamp}_{slug}_raw_calls_{suffix:02d}.md"
            )
            suffix += 1
        self.file_path = candidate
        return self.file_path
    def _write_file(self) -> None:
        path = self._resolve_file_path()
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(_render_markdown(self), encoding="utf-8")
 _active_session: contextvars.ContextVar[Optional[CallLogContext]] = (
    contextvars.ContextVar("_active_session", default=None)
 )
 def active_session() -> Optional[CallLogContext]:
    """Return the currently active :class:`CallLogContext`, if any."""
    return _active_session.get()
@contextmanager
 def session(action: str, out_dir: Path) -> Iterator[CallLogContext]:
    """Open a per-action debug session.
    On entry, sets the module-level ``_active_session`` ContextVar so any
    ``wrap_rpc_http``'d clients on the same thread/task record their calls.
    On exit, the previous active session (if any) is restored.
    """
    ctx = CallLogContext(
        action=str(action),
        out_dir=Path(out_dir),
        started_at=datetime.now(),
    )
    token = _active_session.set(ctx)
    try:
        yield ctx
    finally:
        _active_session.reset(token)
 def wrap_rpc_http(rpc: Any) -> None:
    """Idempotently wrap ``rpc.post`` / ``rpc.get``.
    When a session is active (``_active_session.get() is not None``), the
    wrapped methods perform the HTTP call themselves with ``requests`` and
    record the call before returning the same value ``BaseRequest`` would have
    returned. When no session is active, the wrapped methods delegate to the
    original implementation, preserving stock ``BaseRequest`` behavior.
    Calling this twice on the same instance is a no-op. The wrapper does not
    alter ``rpc.form_post`` (no Sirna action calls it as of plan 3).
    """
    if rpc is None:
        return
    if getattr(rpc, "_debug_call_log_wrapped", False):
        return
    rpc._orig_post = rpc.post
    rpc._orig_get = rpc.get
    def _wrapped_post(
        url: str,
        params: Any = None,
        files: Any = None,
        headers: Optional[dict] = None,
    ) -> Any:
        ctx = _active_session.get()
        if ctx is None:
            kwargs = {}
            if params is not None:
                kwargs["params"] = params
            if files is not None:
                kwargs["files"] = files
            if headers is not None:
                kwargs["headers"] = headers
            return rpc._orig_post(url, **kwargs)
        effective_params = params if params is not None else {}
        effective_headers = (
            headers
            if headers is not None
            else {"Content-Type": "application/json"}
        )
        source = _detect_source(rpc)
        request_body = _redact(effective_params)
        record = CallRecord(
            index=0,
            method="POST",
            url=str(url),
            path=_url_path(url),
            source=source,
            transport=_pick_transport(effective_params),
            http_status=None,
            request_body=request_body,
            response_body=None,
            error=None,
        )
        return_value: Any = None
        try:
            response = requests.post(
                url,
                data=json.dumps(effective_params) if effective_params else None,
                headers=effective_headers,
                timeout=_DEFAULT_TIMEOUT_POST,
                files=files,
            )
        except Exception as exc:  # pragma: no cover - delegated to logger
            record.error = f"transport error: {exc}"
            try:
                rpc.get_logger().error(f"Request ERROR: {exc}")
            except Exception:
                pass
            ctx.append(record)
            return None
        record.http_status = response.status_code
        record.response_body, parse_error = _decode_response_body(response)
        try:
            rpc.get_logger().debug(
                f"Request >>> : {response.request.body} "
                f"{response.status_code} {response.text}"
            )
        except Exception:
            pass
        if response.status_code == 200:
            if parse_error is not None:
                record.error = f"json parse error: {parse_error}"
                return_value = None
            else:
                return_value = record.response_body
        else:
            record.error = f"HTTP {response.status_code}: {response.text}"
            try:
                rpc.get_logger().error(
                    f"Request ERROR: ('Request ERROR:', {response.text!r})"
                )
            except Exception:
                pass
            return_value = None
        ctx.append(record)
        return return_value
    def _wrapped_get(
        url: str,
        params: Any = None,
        headers: Optional[dict] = None,
    ) -> Any:
        ctx = _active_session.get()
        if ctx is None:
            kwargs = {}
            if params is not None:
                kwargs["params"] = params
            if headers is not None:
                kwargs["headers"] = headers
            return rpc._orig_get(url, **kwargs)
        effective_params = params if params is not None else {}
        effective_headers = (
            headers
            if headers is not None
            else {"Content-Type": "application/json"}
        )
        source = _detect_source(rpc)
        request_body = _redact(effective_params)
        record = CallRecord(
            index=0,
            method="GET",
            url=str(url),
            path=_url_path(url),
            source=source,
            transport="params",
            http_status=None,
            request_body=request_body,
            response_body=None,
            error=None,
        )
        return_value: Any = None
        try:
            response = requests.get(
                url,
                params=effective_params,
                headers=effective_headers,
                timeout=_DEFAULT_TIMEOUT_GET,
            )
        except Exception as exc:  # pragma: no cover - delegated to logger
            record.error = f"transport error: {exc}"
            try:
                rpc.get_logger().error(f"Request ERROR: {exc}")
            except Exception:
                pass
            ctx.append(record)
            return None
        record.http_status = response.status_code
        record.response_body, parse_error = _decode_response_body(response)
        try:
            rpc.get_logger().debug(
                f"Request >>> : {effective_params} "
                f"{response.status_code} {response.text}"
            )
        except Exception:
            pass
        if response.status_code == 200:
            if parse_error is not None:
                record.error = f"json parse error: {parse_error}"
                return_value = None
            else:
                return_value = record.response_body
        ctx.append(record)
        return return_value
    rpc.post = _wrapped_post
    rpc.get = _wrapped_get
    rpc._debug_call_log_wrapped = True
 # ---------------------------------------------------------------------------
 # Helpers
 # ---------------------------------------------------------------------------
 _URL_PATH_RE = re.compile(r"https?://[^/]+(/.*)?$")
 _SLUG_RE = re.compile(r"[^A-Za-z0-9._-]+")
 def _slugify_action(action: str) -> str:
    slug = _SLUG_RE.sub("_", str(action)).strip("_")
    return slug or "action"
 def _url_path(url: Any) -> str:
    text = str(url or "")
    match = _URL_PATH_RE.match(text)
    if match and match.group(1):
        return match.group(1)
    if text.startswith("/"):
        return text
    return text
 def _pick_transport(params: Any) -> str:
    if isinstance(params, dict) and "data" in params:
        return "data"
    return "params"
 def _detect_source(rpc: Any) -> str:
    """Walk the call stack to find the outermost frame whose ``self`` is rpc."""
    try:
        stack = inspect.stack()
    except Exception:
        return ""
    candidate = ""
    try:
        for frame_info in stack:
            frame = frame_info.frame
            if frame.f_locals.get("self", None) is rpc:
                candidate = frame_info.function
        return candidate
    finally:
        del stack
 def _redact(params: Any) -> Any:
    """Return a copy of ``params`` with ``apiKey`` redacted."""
    try:
        cloned = copy.deepcopy(params)
    except Exception:
        return params
    _redact_in_place(cloned)
    return cloned
 def _redact_in_place(value: Any) -> None:
    if isinstance(value, dict):
        for key in list(value.keys()):
            if isinstance(key, str) and key.lower() == "apikey":
                value[key] = "<redacted>"
            else:
                _redact_in_place(value[key])
    elif isinstance(value, list):
        for item in value:
            _redact_in_place(item)
 def _decode_response_body(response: Any) -> tuple[Any, Optional[str]]:
    """Best-effort response decoding used for both record + return value."""
    text = getattr(response, "text", "")
    try:
        return response.json(), None
    except Exception as exc:
        if text:
            return {"raw_text": text}, str(exc)
        return None, str(exc)
 # ---------------------------------------------------------------------------
 # Markdown rendering
 # ---------------------------------------------------------------------------
 def _render_markdown(ctx: CallLogContext) -> str:
    title = f"# {ctx.action} Raw Call/Response Log"
    parts: List[str] = [title, ""]
    parts.append("## LIMS Calls")
    parts.append("")
    parts.append("| # | Method | Path | Source | HTTP |")
    parts.append("|---|---|---|---|---|")
    for record in ctx.calls:
        anchor = _row_anchor(record)
        http = (
            f"`{record.http_status}`"
            if record.http_status is not None
            else "`-`"
        )
        parts.append(
            f"| [{record.index}](#{anchor}) | `{record.method}` | "
            f"`{record.path}` | `{record.source}` | {http} |"
        )
    parts.append("")
    for record in ctx.calls:
        parts.append(f"## {record.index} {record.method} {record.path}")
        parts.append("")
        parts.append(f"- Source: `{record.source}`")
        parts.append(f"- Transport: `{record.transport}`")
        if record.http_status is not None:
            parts.append(f"- HTTP status: `{record.http_status}`")
        else:
            parts.append("- HTTP status: `-`")
        if record.error:
            parts.append(f"- Error: {record.error}")
        parts.append("")
        parts.append("### Request Body")
        parts.append("")
        parts.append("```json")
        parts.append(_to_json_block(record.request_body))
        parts.append("```")
        parts.append("")
        parts.append("### Response Body")
        parts.append("")
        parts.append("```json")
        parts.append(_to_json_block(record.response_body))
        parts.append("```")
        parts.append("")
    return "\n".join(parts).rstrip() + "\n"
 def _row_anchor(record: CallRecord) -> str:
    """Build a GitHub-style anchor matching ``## N METHOD /path``."""
    raw = f"{record.index}-{record.method}-{record.path}"
    raw = raw.lower()
    raw = re.sub(r"[^a-z0-9]+", "-", raw)
    return raw.strip("-")
 def _to_json_block(value: Any) -> str:
    try:
        return json.dumps(value, ensure_ascii=False, indent=2, sort_keys=True)
    except TypeError:
        return json.dumps(str(value), ensure_ascii=False, indent=2)
--- a/unilabos/devices/workstation/bioyond_studio/sirna_station/init.py
+++ b/unilabos/devices/workstation/bioyond_studio/sirna_station/init.py
@@ -0,0 +1,3 @@
 from .sirna_station import BioyondSirnaStation, fetch_workflow_list, load_sirna_config
 __all__ = ["BioyondSirnaStation", "fetch_workflow_list", "load_sirna_config"]
--- a/unilabos/devices/workstation/bioyond_studio/sirna_station/sirna_station.py
+++ b/unilabos/devices/workstation/bioyond_studio/sirna_station/sirna_station.py
--- a/unilabos/devices/workstation/bioyond_studio/station.py
+++ b/unilabos/devices/workstation/bioyond_studio/station.py
@@ -7,6 +7,7 @@ Bioyond Workstation Implementation
 import time
 import traceback
 import threading
 from contextlib import contextmanager
 from datetime import datetime
 from typing import Dict, Any, List, Optional, Union
 import json
@@ -14,6 +15,7 @@ from pathlib import Path
 from unilabos.devices.workstation.workstation_base import WorkstationBase, ResourceSynchronizer
 from unilabos.devices.workstation.bioyond_studio.bioyond_rpc import BioyondV1RPC
 from unilabos.devices.workstation.bioyond_studio import debug_call_log
 from unilabos.registry.placeholder_type import ResourceSlot, DeviceSlot
 from unilabos.resources.warehouse import WareHouse
 from unilabos.utils.log import logger
@@ -174,6 +176,8 @@ class BioyondResourceSynchronizer(ResourceSynchronizer):
                logger.warning("从Bioyond获取的物料数据为空")
                return False
            self._update_material_cache_from_stock(all_bioyond_data)
            # 转换为UniLab格式
            unilab_resources = resource_bioyond_to_plr(
                all_bioyond_data,
@@ -187,6 +191,29 @@ class BioyondResourceSynchronizer(ResourceSynchronizer):
            logger.error(f"从Bioyond同步物料数据失败: {e}")
            return False
    def _update_material_cache_from_stock(self, materials: List[Dict[str, Any]]) -> None:
        """用本次库存查询结果同步 RPC 的 name -> material id 缓存。"""
        material_cache = getattr(self.bioyond_api_client, "material_cache", None)
        if not isinstance(material_cache, dict):
            return
        before_count = len(material_cache)
        for material in materials:
            material_name = material.get("name")
            material_id = material.get("id")
            if material_name and material_id:
                material_cache[material_name] = material_id
            for detail_material in material.get("detail", []) or []:
                detail_name = detail_material.get("name")
                detail_id = detail_material.get("detailMaterialId") or detail_material.get("id")
                if detail_name and detail_id:
                    material_cache[detail_name] = detail_id
        logger.debug(
            f"已用Bioyond库存同步物料缓存: {before_count} -> {len(material_cache)}"
        )
    def sync_to_external(self, resource: Any) -> bool:
        """将本地物料数据变更同步到Bioyond系统"""
        try:
@@ -678,6 +705,70 @@ class BioyondWorkstation(WorkstationBase):
    集成Bioyond物料管理的工作站实现
    """
    # 子类（如 sirna / peptide）覆写以指定默认 raw-call 日志目录。
    # 路径相对仓库根；为 None 时若 debug_log=True 仍会写入临时位置。
    _DEBUG_LOG_DEFAULT_DIR: Optional[str] = None
    def _create_bioyond_rpc(self, config: Dict[str, Any]) -> BioyondV1RPC:
        """创建 Bioyond RPC 客户端并应用调试包装。
        所有创建 ``BioyondV1RPC`` 的路径（饿汉初始化、Sirna 延迟初始化、
        以及未来的前端重新配置路径）都应通过该 helper，
        以确保 debug_log 包装与命名/日志策略保持一致。
        """
        rpc = BioyondV1RPC(config)
        debug_call_log.wrap_rpc_http(rpc)
        return rpc
    def _set_hardware_interface(self, rpc: BioyondV1RPC) -> BioyondV1RPC:
        """将已构造的 RPC 客户端设置到 ``self.hardware_interface``，并应用调试包装。"""
        debug_call_log.wrap_rpc_http(rpc)
        self.hardware_interface = rpc
        return rpc
    def _debug_log_resolved_dir(self) -> Path:
        """解析 ``debug_log_dir`` 为绝对路径。"""
        configured = (getattr(self, "bioyond_config", {}) or {}).get("debug_log_dir")
        default_dir = getattr(self, "_DEBUG_LOG_DEFAULT_DIR", None)
        candidate = configured or default_dir or "temp_benyao/_logs/bioyond_debug"
        path = Path(candidate)
        if not path.is_absolute():
            repo_root = Path(__file__).resolve().parents[4]
            path = repo_root / path
        return path
    def _ensure_debug_log_state(self) -> None:
        """从 ``self.bioyond_config`` 派生 ``_debug_log_enabled`` / ``_debug_log_dir``。
        每次进入 ``_debug_call_session`` 时都重新解析，以兼容前端在运行时
        修改 ``bioyond_config['debug_log']`` 或目录的场景；同时也容忍
        子类（如 Sirna 延迟初始化）在 ``__init__`` 早期未触发本方法。
        """
        cfg = getattr(self, "bioyond_config", {}) or {}
        self._debug_log_enabled = bool(cfg.get("debug_log"))
        self._debug_log_dir = self._debug_log_resolved_dir()
    @contextmanager
    def _debug_call_session(self, action_name: str):
        """在 action 体外加一层 debug 会话上下文。
        - ``debug_log`` 关闭时是空上下文，开销为 0。
        - ``debug_log`` 开启时进入 :func:`debug_call_log.session`，所有
          已被 ``wrap_rpc_http`` 包装过的 RPC 客户端都会捕获本次 action
          产生的 HTTP 调用并写入 Markdown 文件。
        子类（如 ``end_experiment``、``manual_unload`` 等）可以直接在
        action 体里以 ``with self._debug_call_session("action_name"):`` 包裹。
        """
        cfg = getattr(self, "bioyond_config", {}) or {}
        enabled = bool(cfg.get("debug_log"))
        if not enabled:
            yield None
            return
        out_dir = BioyondWorkstation._debug_log_resolved_dir(self)
        with debug_call_log.session(action_name, out_dir) as ctx:
            yield ctx
    def _publish_task_status(
        self,
        task_id: str,
@@ -862,7 +953,7 @@ class BioyondWorkstation(WorkstationBase):
            self.bioyond_config = {}
            print("警告: 未提供 bioyond_config，请确保在 JSON 配置文件中提供完整配置")
-        self.hardware_interface = BioyondV1RPC(self.bioyond_config)
+        self.hardware_interface = self._create_bioyond_rpc(self.bioyond_config)
    def resource_tree_add(self, resources: List[ResourcePLR]) -> None:
        """添加资源到资源树并更新ROS节点
@@ -1338,11 +1429,7 @@ class BioyondWorkstation(WorkstationBase):
            if self.hardware_interface:
                self.hardware_interface.scheduler_reset()
-            # 刷新物料缓存
+            # 重新同步资源，并用同一次库存查询结果更新物料缓存
            if self.hardware_interface:
                self.hardware_interface.refresh_material_cache()
            # 重新同步资源
            if self.resource_synchronizer:
                self.resource_synchronizer.sync_from_external()
--- a/unilabos/resources/bioyond/init.py
+++ b/unilabos/resources/bioyond/init.py
@@ -0,0 +1 @@
 from . import sirna_materials  # noqa: F401  ensure @resource classes are importable for PLR deserialize
--- a/unilabos/resources/bioyond/decks.py
+++ b/unilabos/resources/bioyond/decks.py
@@ -1,6 +1,8 @@
 from os import name
 from pylabrobot.resources import Deck, Coordinate, Rotation
 from unilabos.registry.decorators import resource
 from unilabos.resources.bioyond.YB_warehouses import (
    bioyond_warehouse_1x4x4,
    bioyond_warehouse_1x4x4_right,  # 新增：右侧仓库 (A05～D08)
@@ -23,6 +25,11 @@ from unilabos.resources.bioyond.YB_warehouses import (
 from unilabos.resources.bioyond.warehouses import (
    bioyond_warehouse_tipbox_storage_left,   # 新增：Tip盒堆栈(左)
    bioyond_warehouse_tipbox_storage_right,  # 新增：Tip盒堆栈(右)
    bioyond_warehouse_sirna_automation_stack,
    bioyond_warehouse_sirna_centrifuge_balance_plate_stack,
    bioyond_warehouse_sirna_g3_liquid_handler,
    bioyond_warehouse_numeric_stack,  # 新增：数字编码堆栈 (用于多肽站)
    bioyond_warehouse_live_grid,
 )
@@ -101,6 +108,83 @@ class BIOYOND_PolymerPreparationStation_Deck(Deck):
        for warehouse_name, warehouse in self.warehouses.items():
            self.assign_child_resource(warehouse, location=self.warehouse_locations[warehouse_name])
@resource(
    id="BIOYOND_SirnaStation_Deck",
    category=["deck"],
    description="BIOYOND 小核酸工作站 Deck",
    icon="配液站.webp",
 )
 class BIOYOND_SirnaStation_Deck(Deck):
    WAREHOUSE_BIOYOND_AXIS = {
        "G3移液站": "xy_col_row",
        "自动化堆栈": "xy_col_row",
        "离心机配平板堆栈": "xy_col_row",
    }
    WAREHOUSE_BIOYOND_KEY_AXIS = {
        "G3移液站": "col_row",
        "自动化堆栈": "col_row",
        "离心机配平板堆栈": "col_row",
    }
    # Bioyond warehouse UUID -> 本地仓库名称 映射。
    # 留空时由配置（station config 的 ``warehouse_bioyond_ids``）注入。
    # graph 节点也可在 deck.config.warehouse_bioyond_ids 覆盖。
    WAREHOUSE_BIOYOND_IDS: dict = {}
    def __init__(
        self,
        name: str = "SirnaStation_Deck",
        size_x: float = 2700.0,
        size_y: float = 1080.0,
        size_z: float = 1500.0,
        category: str = "deck",
        setup: bool = False,
        warehouse_bioyond_ids: dict | None = None,
        **kwargs,
    ) -> None:
        super().__init__(name=name, size_x=size_x, size_y=size_y, size_z=size_z)
        # 按需写入实例级覆盖；保留默认空 mapping，避免改动模型常量。
        self.warehouse_bioyond_ids: dict = dict(self.WAREHOUSE_BIOYOND_IDS)
        if warehouse_bioyond_ids:
            self.warehouse_bioyond_ids.update(warehouse_bioyond_ids)
        if setup:
            self.setup()
    @classmethod
    def deserialize(cls, data: dict, allow_marshal: bool = False):
        if data.get("children") and data.get("setup") is True:
            data = data.copy()
            data["setup"] = False
        result = super().deserialize(data, allow_marshal=allow_marshal)
        result._ensure_sirna_warehouse_metadata()
        return result
    def _ensure_sirna_warehouse_metadata(self) -> None:
        for child in getattr(self, "children", []):
            name = getattr(child, "name", "")
            axis = self.WAREHOUSE_BIOYOND_AXIS.get(name)
            if axis and not hasattr(child, "bioyond_axis"):
                child.bioyond_axis = axis
            key_axis = self.WAREHOUSE_BIOYOND_KEY_AXIS.get(name)
            if key_axis and not hasattr(child, "bioyond_key_axis"):
                child.bioyond_key_axis = key_axis
    def setup(self) -> None:
        # Sirna 读接口 /api/storage/location/locations-by-type 返回完整固定堆栈清单。
        # LIMS 在库物料接口仍使用相同的 自动化堆栈 名称和数字库位编码。
        self.warehouses = {
            "G3移液站": bioyond_warehouse_sirna_g3_liquid_handler(),
            "自动化堆栈": bioyond_warehouse_sirna_automation_stack(),
            "离心机配平板堆栈": bioyond_warehouse_sirna_centrifuge_balance_plate_stack(),
        }
        self.warehouse_locations = {
            "G3移液站": Coordinate(0.0, 0.0, 0.0),
            "自动化堆栈": Coordinate(220.0, 0.0, 0.0),
            "离心机配平板堆栈": Coordinate(1740.0, 0.0, 0.0),
        }
        for warehouse_name, warehouse in self.warehouses.items():
            self.assign_child_resource(warehouse, location=self.warehouse_locations[warehouse_name])
 class BIOYOND_YB_Deck(Deck):
    def __init__(
        self,
@@ -150,12 +234,146 @@ class BIOYOND_YB_Deck(Deck):
        for warehouse_name, warehouse in self.warehouses.items():
            self.assign_child_resource(warehouse, location=self.warehouse_locations[warehouse_name])
@resource(
    id="BIOYOND_PeptideStation_Deck",
    category=["deck"],
    description="BIOYOND 多肽工作站 Deck",
    icon="preparation_station.webp",
 )
 class BIOYOND_PeptideStation_Deck(Deck):
    WAREHOUSE_BIOYOND_AXIS = dict.fromkeys(
        [
            "自动化堆栈",
            "低温冰箱仓库",
            "Tecan移液站库",
            "G3移液站库",
            "IDOT移液站库",
            "G3缓冲库",
            "盖板缓冲库",
            "配平板缓冲库",
            "IDOT缓冲库",
            "固相合成板底座缓冲位",
            "离心机库位",
            "热封膜机位",
        ],
        "xy_col_row",
    )
    WAREHOUSE_BIOYOND_KEY_AXIS = dict.fromkeys(WAREHOUSE_BIOYOND_AXIS, "row_col")
    def __init__(
        self,
        name: str = "PeptideStation_Deck",
        size_x: float = 3500.0,
        size_y: float = 1800.0,
        size_z: float = 1500.0,
        category: str = "deck",
        setup: bool = False
    ) -> None:
        super().__init__(name=name, size_x=size_x, size_y=size_y, size_z=size_z)
        if setup:
            self.setup()
    @classmethod
    def deserialize(cls, data: dict, allow_marshal: bool = False):
        if data.get("children") and data.get("setup") is True:
            # 已有序列化子资源，跳过 setup 避免重复创建
            result = super(BIOYOND_PeptideStation_Deck, cls).deserialize(data, allow_marshal=allow_marshal)
        else:
            result = super(BIOYOND_PeptideStation_Deck, cls).deserialize(data, allow_marshal=allow_marshal)
        result._ensure_peptide_warehouse_metadata()
        return result
    def _ensure_peptide_warehouse_metadata(self) -> None:
        for child in getattr(self, "children", []):
            name = getattr(child, "name", "")
            axis = self.WAREHOUSE_BIOYOND_AXIS.get(name)
            if axis and not hasattr(child, "bioyond_axis"):
                child.bioyond_axis = axis
            key_axis = self.WAREHOUSE_BIOYOND_KEY_AXIS.get(name)
            if key_axis and not hasattr(child, "bioyond_key_axis"):
                child.bioyond_key_axis = key_axis
    def setup(self) -> None:
        # 多肽工作站仓库配置
        # 基于 2026-05-09 live API probe 发现的实际仓库拓扑 (12个仓库)
        # 数据来源: temp_benyao/peptide/_logs/warehouse_discovery_raw_live_2026-05-09.json
        self.warehouses = {
            # 主自动化堆栈 - live API: code 10-17 -> x=17, y=10，显示为 10 行×17 列
            "自动化堆栈": bioyond_warehouse_numeric_stack(
                "自动化堆栈", rows=10, columns=17, bioyond_axis="xy_col_row", bioyond_key_axis="row_col"
            ),
            # 低温存储
            "低温冰箱仓库": bioyond_warehouse_live_grid(
                "低温冰箱仓库", rows=2, columns=3, slot_keys=["1", "2", "3", "4", "5", "6"]
            ),
            # 移液站库位
            "Tecan移液站库": bioyond_warehouse_live_grid(
                "Tecan移液站库", rows=1, columns=18, slot_keys=[str(index) for index in range(1, 19)]
            ),
            "G3移液站库": bioyond_warehouse_live_grid(
                "G3移液站库",
                rows=1,
                columns=18,
                slot_keys=["1", "2", "3", "4", "垃圾桶", "6", "7", "8", "9", "10", "11", "12", "13", "14", "15", "16", "17", "18"],
            ),
            "IDOT移液站库": bioyond_warehouse_live_grid(
                "IDOT移液站库",
                rows=1,
                columns=12,
                slot_keys=[f"0009-{index:04d}" for index in range(1, 13)],
            ),
            # 缓冲库位
            "G3缓冲库": bioyond_warehouse_live_grid(
                "G3缓冲库", rows=1, columns=5, slot_keys=[str(index) for index in range(1, 6)]
            ),
            "盖板缓冲库": bioyond_warehouse_live_grid(
                "盖板缓冲库", rows=1, columns=7, slot_keys=[str(index) for index in range(1, 8)]
            ),
            "配平板缓冲库": bioyond_warehouse_live_grid(
                "配平板缓冲库", rows=1, columns=3, slot_keys=[str(index) for index in range(1, 4)]
            ),
            "IDOT缓冲库": bioyond_warehouse_live_grid(
                "IDOT缓冲库", rows=1, columns=2, slot_keys=["1", "1"]
            ),
            "固相合成板底座缓冲位": bioyond_warehouse_live_grid(
                "固相合成板底座缓冲位",
                rows=1,
                columns=4,
                slot_keys=[f"0015-{index:04d}" for index in range(1, 5)],
            ),
            # 设备库位
            "离心机库位": bioyond_warehouse_live_grid(
                "离心机库位", rows=1, columns=4, slot_keys=[f"0017-{index:04d}" for index in range(1, 5)]
            ),
            "热封膜机位": bioyond_warehouse_live_grid(
                "热封膜机位", rows=1, columns=2, slot_keys=[f"0016-{index:04d}" for index in range(1, 3)]
            ),
        }
        # 仓库位置布局 (需根据实际硬件布局调整)
        self.warehouse_locations = {
            "自动化堆栈": Coordinate(0.0, 0.0, 0.0),
            "Tecan移液站库": Coordinate(0.0, 1150.0, 0.0),
            "G3移液站库": Coordinate(0.0, 1300.0, 0.0),
            "IDOT移液站库": Coordinate(0.0, 1450.0, 0.0),
            "G3缓冲库": Coordinate(0.0, 1600.0, 0.0),
            "盖板缓冲库": Coordinate(850.0, 1600.0, 0.0),
            "低温冰箱仓库": Coordinate(2700.0, 0.0, 0.0),
            "配平板缓冲库": Coordinate(2700.0, 300.0, 0.0),
            "IDOT缓冲库": Coordinate(2700.0, 450.0, 0.0),
            "固相合成板底座缓冲位": Coordinate(2700.0, 600.0, 0.0),
            "离心机库位": Coordinate(2700.0, 750.0, 0.0),
            "热封膜机位": Coordinate(2700.0, 900.0, 0.0),
        }
        for warehouse_name, warehouse in self.warehouses.items():
            self.assign_child_resource(warehouse, location=self.warehouse_locations[warehouse_name])
 def YB_Deck(name: str) -> Deck:
    by=BIOYOND_YB_Deck(name=name)
    by.setup()
    return by
--- a/unilabos/resources/bioyond/sirna_materials.py
+++ b/unilabos/resources/bioyond/sirna_materials.py
@@ -0,0 +1,126 @@
 """Sirna Station Material Resource Definitions
 Defines PyLabRobot resource classes for Bioyond Sirna station materials.
 Each class is decorated with @resource for AST-based registry discovery.
 """
 from collections import OrderedDict
 from pylabrobot.resources import Plate, TipRack, Container
 from unilabos.registry.decorators import resource
@resource(
    id="bioyond_sirna_g3_200ul_tip_rack",
    category=["labware", "tip_rack"],
    description="G3-200ul枪头盒 for Sirna station",
 )
 class BioyondSirna_G3_200ul_TipRack(TipRack):
    """G3-200ul tip rack for Sirna liquid handling."""
    def __init__(self, *args, **kwargs):
        kwargs.setdefault("size_x", 127.76)
        kwargs.setdefault("size_y", 85.48)
        kwargs.setdefault("size_z", 64.0)
        kwargs.setdefault("model", "bioyond_sirna_g3_200ul_tip_rack")
        kwargs.setdefault("with_tips", True)
        if kwargs.get("ordering") is None and kwargs.get("ordered_items") is None:
            kwargs["ordering"] = OrderedDict()
        super().__init__(*args, **kwargs)
@resource(
    id="bioyond_sirna_g3_50ul_tip_rack",
    category=["labware", "tip_rack"],
    description="G3-50ul枪头盒 for Sirna station",
 )
 class BioyondSirna_G3_50ul_TipRack(TipRack):
    """G3-50ul tip rack for Sirna liquid handling."""
    def __init__(self, *args, **kwargs):
        kwargs.setdefault("size_x", 127.76)
        kwargs.setdefault("size_y", 85.48)
        kwargs.setdefault("size_z", 64.0)
        kwargs.setdefault("model", "bioyond_sirna_g3_50ul_tip_rack")
        kwargs.setdefault("with_tips", True)
        if kwargs.get("ordering") is None and kwargs.get("ordered_items") is None:
            kwargs["ordering"] = OrderedDict()
        super().__init__(*args, **kwargs)
@resource(
    id="bioyond_sirna_384_well_plate",
    category=["labware", "plate"],
    description="384孔板 for Sirna assays",
 )
 class BioyondSirna_384WellPlate(Plate):
    """384-well plate for Sirna reporter gene detection."""
    def __init__(self, *args, **kwargs):
        kwargs.setdefault("size_x", 127.76)
        kwargs.setdefault("size_y", 85.48)
        kwargs.setdefault("size_z", 14.35)
        kwargs.setdefault("model", "bioyond_sirna_384_well_plate")
        kwargs.setdefault("plate_type", "skirted")
        if kwargs.get("ordering") is None and kwargs.get("ordered_items") is None:
            kwargs["ordering"] = OrderedDict()
        super().__init__(*args, **kwargs)
@resource(
    id="bioyond_sirna_cell_culture_plate",
    category=["labware", "plate"],
    description="细胞培养板 for Sirna cell culture",
 )
 class BioyondSirna_CellCulturePlate(Plate):
    """Cell culture plate for Sirna experiments."""
    def __init__(self, *args, **kwargs):
        kwargs.setdefault("size_x", 127.76)
        kwargs.setdefault("size_y", 85.48)
        kwargs.setdefault("size_z", 14.35)
        kwargs.setdefault("model", "bioyond_sirna_cell_culture_plate")
        kwargs.setdefault("plate_type", "skirted")
        if kwargs.get("ordering") is None and kwargs.get("ordered_items") is None:
            kwargs["ordering"] = OrderedDict()
        super().__init__(*args, **kwargs)
@resource(
    id="bioyond_sirna_reagent_trough",
    category=["labware", "trough"],
    description="试剂槽 for Sirna reagents",
 )
 class BioyondSirna_ReagentTrough(Container):
    """Reagent trough for Sirna station reagents (RiboGreen, etc.)."""
    def __init__(self, *args, **kwargs):
        kwargs.setdefault("size_x", 127.76)
        kwargs.setdefault("size_y", 85.48)
        kwargs.setdefault("size_z", 44.0)
        kwargs.setdefault("max_volume", 300000.0)
        kwargs.setdefault("model", "bioyond_sirna_reagent_trough")
        super().__init__(*args, **kwargs)
 # Material type code mapping for dynamic instantiation
 MATERIAL_TYPE_CODE_TO_CLASS = {
    "0016": BioyondSirna_G3_200ul_TipRack,
    "0017": BioyondSirna_G3_50ul_TipRack,
    "0015": BioyondSirna_384WellPlate,
    "0001": BioyondSirna_CellCulturePlate,
    "0006": BioyondSirna_ReagentTrough,
 }
 def get_material_class_by_type_code(type_code: str):
    """Get resource class by Bioyond material type code.
    Args:
        type_code: Bioyond materialTypeCode (e.g., "0016", "0017")
    Returns:
        Resource class or None if not found
    """
    return MATERIAL_TYPE_CODE_TO_CLASS.get(type_code)
--- a/unilabos/resources/bioyond/warehouses.py
+++ b/unilabos/resources/bioyond/warehouses.py
@@ -1,5 +1,180 @@
 from pylabrobot.resources import Coordinate
 from pylabrobot.resources.carrier import ResourceHolder, create_homogeneous_resources
 from unilabos.resources.warehouse import WareHouse, warehouse_factory
 class BioyondWareHouse(WareHouse):
    """Bioyond 仓库，额外保存服务端 x/y 坐标和库位标签语义。"""
    def __init__(self, *args, bioyond_axis: str = "xy_row_col", bioyond_key_axis: str = "row_col", **kwargs):
        super().__init__(*args, **kwargs)
        self.bioyond_axis = bioyond_axis
        self.bioyond_key_axis = bioyond_key_axis
    def serialize(self) -> dict:
        data = super().serialize()
        data["bioyond_axis"] = self.bioyond_axis
        data["bioyond_key_axis"] = self.bioyond_key_axis
        return data
 def bioyond_warehouse_numeric_stack(
    name: str,
    rows: int = 10,
    columns: int = 17,
    bioyond_axis: str = "xy_row_col",
    bioyond_key_axis: str = "row_col",
 ) -> WareHouse:
    """创建 Bioyond 数字库位堆栈，库位名使用服务端返回的 行-列 格式。
    bioyond_axis: 仓库级别的 Bioyond 坐标轴约定，供 graphio 的坐标映射使用。
        - "xy_row_col" (default): Bioyond x→row, y→col (reaction/peptide 历史约定).
        - "xy_col_row": Bioyond x→col, y→row (Sirna live API 实测约定).
    bioyond_key_axis: 库位标签生成约定。
        - "row_col" (default): 视觉行列和标签行列一致，例如 10 行 x 17 列 → 1-1..10-17。
        - "col_row": 视觉行列转置，但标签仍保持 Bioyond row-col，例如
          17 行 x 10 列 → 1-1..10-17。
    未设置时 graphio 回退到默认 "xy_row_col"，其他调用方保持原行为。
    """
    num_items_x = columns
    num_items_y = rows
    num_items_z = 1
    dx = 10.0
    dy = 10.0
    dz = 10.0
    item_dx = 147.0
    item_dy = 106.0
    item_dz = 130.0
    locations = [
        Coordinate(dx + col * item_dx, dy + row * item_dy, dz)
        for row in range(num_items_y)
        for col in range(num_items_x)
    ]
    holders = create_homogeneous_resources(
        klass=ResourceHolder,
        locations=locations,
        resource_size_x=127.0,
        resource_size_y=86.0,
        resource_size_z=25.0,
        name_prefix=name,
    )
    if bioyond_key_axis == "row_col":
        keys = [
            f"{row + 1}-{col + 1}"
            for row in range(num_items_y)
            for col in range(num_items_x)
        ]
    elif bioyond_key_axis == "col_row":
        keys = [
            f"{col + 1}-{row + 1}"
            for row in range(num_items_y)
            for col in range(num_items_x)
        ]
    else:
        raise ValueError(f"未知 Bioyond 库位标签约定: {bioyond_key_axis!r}")
    warehouse = BioyondWareHouse(
        name=name,
        size_x=dx + item_dx * num_items_x,
        size_y=dy + item_dy * num_items_y,
        size_z=dz + item_dz * num_items_z,
        num_items_x=num_items_x,
        num_items_y=num_items_y,
        num_items_z=num_items_z,
        ordering_layout="row-major",
        sites={key: holder for key, holder in zip(keys, holders.values())},
        category="warehouse",
        bioyond_axis=bioyond_axis,
        bioyond_key_axis=bioyond_key_axis,
    )
    return warehouse
 def bioyond_warehouse_live_grid(
    name: str,
    rows: int,
    columns: int,
    slot_keys: list[str] | None = None,
    bioyond_axis: str = "xy_col_row",
    bioyond_key_axis: str = "row_col",
 ) -> WareHouse:
    """创建 Bioyond 实测库位网格，按服务端 code 保存位点标签。
    默认用于 Peptide live API 返回的坐标：x 是视觉列，y 是视觉行。
    当服务端 code 重复时，为保持 PLR ordering 唯一性，会给后续重复项追加 ``#N``。
    """
    num_items_x = columns
    num_items_y = rows
    num_items_z = 1
    dx = 10.0
    dy = 10.0
    dz = 10.0
    item_dx = 147.0
    item_dy = 106.0
    item_dz = 130.0
    locations = [
        Coordinate(dx + col * item_dx, dy + row * item_dy, dz)
        for row in range(num_items_y)
        for col in range(num_items_x)
    ]
    holders = create_homogeneous_resources(
        klass=ResourceHolder,
        locations=locations,
        resource_size_x=127.0,
        resource_size_y=86.0,
        resource_size_z=25.0,
        name_prefix=name,
    )
    keys = slot_keys or [str(index + 1) for index in range(num_items_x * num_items_y)]
    if len(keys) != len(holders):
        raise ValueError(f"{name} 库位数量不匹配: keys={len(keys)}, holders={len(holders)}")
    seen: dict[str, int] = {}
    unique_keys: list[str] = []
    for key in keys:
        count = seen.get(key, 0) + 1
        seen[key] = count
        unique_keys.append(key if count == 1 else f"{key}#{count}")
    return BioyondWareHouse(
        name=name,
        size_x=dx + item_dx * num_items_x,
        size_y=dy + item_dy * num_items_y,
        size_z=dz + item_dz * num_items_z,
        num_items_x=num_items_x,
        num_items_y=num_items_y,
        num_items_z=num_items_z,
        ordering_layout="row-major",
        sites={key: holder for key, holder in zip(unique_keys, holders.values())},
        category="warehouse",
        bioyond_axis=bioyond_axis,
        bioyond_key_axis=bioyond_key_axis,
    )
 # ================ 小核酸工作站相关堆栈 ================
 def bioyond_warehouse_sirna_g3_liquid_handler(name: str = "G3移液站") -> WareHouse:
    """创建小核酸 G3 移液站库位堆栈：显示为 14 行 x 1 列，标签保持 1-1..1-14。"""
    return bioyond_warehouse_numeric_stack(
        name, rows=14, columns=1, bioyond_axis="xy_col_row", bioyond_key_axis="col_row"
    )
 def bioyond_warehouse_sirna_automation_stack(name: str = "自动化堆栈") -> WareHouse:
    """创建小核酸自动化堆栈：显示为 17 行 x 10 列，标签保持 1-1..10-17。"""
    return bioyond_warehouse_numeric_stack(
        name, rows=17, columns=10, bioyond_axis="xy_col_row", bioyond_key_axis="col_row"
    )
 def bioyond_warehouse_sirna_centrifuge_balance_plate_stack(name: str = "离心机配平板堆栈") -> WareHouse:
    """创建小核酸离心机配平板堆栈：显示为 1 行 x 2 列，标签保持 1-1、2-1。"""
    return bioyond_warehouse_numeric_stack(
        name, rows=1, columns=2, bioyond_axis="xy_col_row", bioyond_key_axis="col_row"
    )
 # ================ 反应站相关堆栈 ================
 def bioyond_warehouse_1x4x4(name: str) -> WareHouse:
--- a/unilabos/resources/graphio.py
+++ b/unilabos/resources/graphio.py
@@ -736,7 +736,7 @@ def resource_bioyond_to_plr(bioyond_materials: list[dict], type_mapping: Dict[st
            logger.warning(f"物料 {unique_name} 不是有效的 ResourcePLR 实例，类型: {type(plr_material)}")
            continue
-        plr_material.code = material.get("code", "") and material.get("barCode", "") or ""
+        plr_material.code = material.get("barCode") or material.get("code") or ""
        plr_material.unilabos_uuid = str(uuid.uuid4())
        # ⭐ 保存 Bioyond 原始信息到 unilabos_extra（用于出库时查询）
@@ -864,11 +864,22 @@ def resource_bioyond_to_plr(bioyond_materials: list[dict], type_mapping: Dict[st
                    warehouse = deck.warehouses[wh_name]
                    logger.debug(f"[Warehouse匹配] 找到warehouse: {wh_name} (容量: {warehouse.capacity}, 行×列: {warehouse.num_items_x}×{warehouse.num_items_y})")
-                    # Bioyond坐标映射 (重要！): x→行(1=A,2=B...), y→列(1=01,2=02...), z→层(通常=1)
+                    # Bioyond坐标映射:
-                    x = loc.get("x", 1)  # 行号 (1-based: 1=A, 2=B, 3=C, 4=D)
+                    # - 历史 row_col 仓库中 x/y 直接按行/列参与索引。
-                    y = loc.get("y", 1)  # 列号 (1-based: 1=01, 2=02, 3=03...)
+                    # - Sirna 的库位标签为 col-row，stock-material 返回 x=标签第二段、y=标签第一段。
                    #   因此 x=13,y=4 应落到 key=4-13，而不是交换后落到 3-5。
                    x = loc.get("x", 1)
                    y = loc.get("y", 1)
                    z = loc.get("z", 1)  # 层号 (1-based, 通常为1)
                    # 仓库级别的轴约定覆盖。
                    # 对旧的 row-col 视觉标签，bioyond_axis="xy_col_row" 需要交换 x/y。
                    # 对 Sirna 的 col-row 库位标签，原始 x/y 已能直接索引到 code 对应位置，不再交换。
                    bioyond_axis = getattr(warehouse, "bioyond_axis", "xy_row_col")
                    bioyond_key_axis = getattr(warehouse, "bioyond_key_axis", "row_col")
                    if bioyond_axis == "xy_col_row" and bioyond_key_axis != "col_row":
                        x, y = y, x
                    # 如果是右侧堆栈，需要调整列号 (5→1, 6→2, 7→3, 8→4)
                    if wh_name == "堆栈1右":
                        y = y - 4  # 将5-8映射到1-4
@@ -912,10 +923,43 @@ def resource_bioyond_to_plr(bioyond_materials: list[dict], type_mapping: Dict[st
                            logger.debug(f"列优先warehouse {wh_name}: x={x}(行),y={y}(列) → row={row_idx},col={col_idx} → idx={idx}")
                    if 0 <= idx < warehouse.capacity:
-                        if warehouse[idx] is None or isinstance(warehouse[idx], ResourceHolder):
+                        slot_key = None
                        ordering = getattr(warehouse, "_ordering", {})
                        sites = getattr(warehouse, "sites", [])
                        if isinstance(ordering, dict) and idx < len(sites):
                            site_at_idx = sites[idx]
                            slot_key = next(
                                (key for key, site in ordering.items() if site is site_at_idx),
                                None,
                            )
                        current_resource = warehouse[idx]
                        if current_resource is None or isinstance(current_resource, (ResourceHolder, str)):
                            if isinstance(current_resource, str):
                                logger.warning(
                                    f"⚠️ 物料 {unique_name} 覆盖 {wh_name}[{idx}]"
                                    f"{f'({slot_key})' if slot_key else ''} 的旧占位 occupied_by={current_resource!r}"
                                )
                            # 物料尺寸已在放入warehouse前根据需要进行了交换
                            warehouse[idx] = plr_material
-                            logger.debug(f"✅ 物料 {unique_name} 放置到 {wh_name}[{idx}] (Bioyond坐标: x={loc.get('x')}, y={loc.get('y')})")
+                            logger.debug(
                                f"✅ 物料 {unique_name} 放置到 {wh_name}[{idx}]"
                                f"{f'({slot_key})' if slot_key else ''} "
                                f"(Bioyond坐标: x={loc.get('x')}, y={loc.get('y')})"
                            )
                        else:
                            parent = getattr(current_resource, "parent", None)
                            current_repr = repr(current_resource)
                            current_len = len(current_resource) if isinstance(current_resource, str) else None
                            logger.warning(
                                f"⚠️ 物料 {unique_name} 跳过放置到 {wh_name}[{idx}]"
                                f"{f'({slot_key})' if slot_key else ''}：目标库位已有 "
                                f"{type(current_resource).__name__}"
                                f"(value={current_repr}, len={current_len})"
                                f"(name={getattr(current_resource, 'name', None)}, "
                                f"parent={getattr(parent, 'name', None)}, "
                                f"uuid={getattr(current_resource, 'unilabos_uuid', None)})"
                            )
                    else:
                        logger.warning(f"❌ 物料 {unique_name} 的索引 {idx} 超出仓库 {wh_name} 容量 {warehouse.capacity}")
                else:
Author	SHA1	Message	Date
yxz321	45b9fd5262	plan: add resource system guidance	2026-05-20 19:17:48 +08:00
yxz321	a1c0b83490	fix: change reset and submit_experiment* from manual confirm to normal node	2026-05-15 15:25:54 +08:00
yxz321	570d6763c0	Fix: update sirna_station to support new flexible resultTable	2026-05-15 14:01:34 +08:00
yxz321	633c8b3d2c	feat: RNA add Sirna experiment controls and workflow bindings - Add bound Sirna workflow names for experiment 1 and experiment 2 submissions. - Route named workflow submission through a shared create-order core with minimal param payloads. - Add direct scheduler controls and cancel/take-out manual confirmation handling.	2026-05-12 19:35:39 +08:00
yxz321	e6ee6fc964	fix: sync Bioyond materials over stale slot placeholders	2026-05-12 19:35:39 +08:00
yxz321	765342c4ff	feat: RNA aggregate Bioyond LIMS reports - Add get_order_report and frontend-like get_aggregated_order_report actions for Sirna LIMS debugging. - Preserve raw checked-section LIMS responses, including collapsed error envelopes, while allowing partial aggregate success. - Add order-id matching safeguards, aggregate report tests, and scaffold assertions for the new manual-confirm actions. - Keep Sirna graph debug logging enabled for report/live-probe troubleshooting.	2026-05-12 19:35:39 +08:00
yxz321	3fc94c6720	feat: RNA refine Sirna material sync and placement - Publish resource tree updates after shared Bioyond external material sync succeeds. - Keep start_experiment scheduler startup non-blocking while reporting unchecked manual load gates. - Prefer Bioyond barCode as PLR material code and add safer occupied-slot diagnostics for warehouse placement.	2026-05-12 19:35:39 +08:00
yxz321	d5f0bca643	fix: RNA use preexisting sync_from_external() from base class. Leave _register_materials_to_tree() commented	2026-05-12 19:35:39 +08:00
yxz321	de51b19e88	fix: RNA restrore manual-confirm table	2026-05-12 19:35:39 +08:00
yxz321	6b94bdd2da	fix: align Bioyond deck warehouse axes - Preserve Sirna col-row labels while flipping visual stack dimensions. - Rebuild Peptide deck warehouses from live API slot geometry and avoid initial graph overlap. - Add Peptide deck layout tests and keep Sirna resource tests passing.	2026-05-12 19:35:39 +08:00
yxz321	d009863c8c	feat: RNA add guided manual unload end_experiment action - Add end_experiment manual_confirm action mirroring start_experiment, with three boolean operator gates and twelve EXECUTOR sibling-array output handles for unloaded material manifests. - Add helpers _build_unload_materials_by_type, _classify_labware_mode, _iter_reagent_liquids, and _clear_unloaded_materials. - Clear unloaded slots and zero reagent liquid contents on confirmation while preserving trough labware; publish single resource tree update after mutations. - Wrap action body in _debug_call_session("end_experiment") for opt-in raw call capture.	2026-05-12 19:35:39 +08:00
yxz321	cae828ce74	feat: RNA add per-action raw HTTP call log for Bioyond station - Add debug_call_log module: contextvar-scoped session(), idempotent wrap_rpc_http, lazy markdown writer with apiKey redaction, source attribution via stack walk, raw_text fallback. - Centralize Bioyond RPC creation in BioyondWorkstation via _create_bioyond_rpc / _set_hardware_interface / _debug_call_session helpers and _DEBUG_LOG_DEFAULT_DIR. - Wrap reset, submit_experiment_1, start_experiment, get_order_list action bodies in _debug_call_session for opt-in per-action capture. - ConnectionMonitor polling stays outside debug sessions and is not logged.	2026-05-12 19:35:39 +08:00
yxz321	5b9f77e81f	feat: RNA simplify station action signatures and add order query - Add explicit reset_operations default; keep hidden plumbing kwargs-only. - Add read-only get_order_list manual_confirm action that calls order_query with latest_only and exposes order_id/order_ids handles. - Collapse start_experiment visible signature to (order_id, materials_loaded, timeout_seconds, assignee_user_ids); legacy params consumed from kwargs to preserve runtime contract. - Reduce submit_experiment_1 graph handles to order_id, resource, coin_cell_code, mount_resource; result dict gains order_id while keeping existing keys.	2026-05-12 19:35:39 +08:00
yxz321	7c83e1bd51	feat: RNA land resource-system mega plan Phases 1-4 plus Phase 5 stack orientation - Phase 1: fix _publish_resource_tree_update to call update_resource via run_async_func with deck resource list. - Phase 2: add ID-first material placement resolver chain with material_info and warehouse_inventory caches. - Phase 3: classify stock-material rows as slot_labware vs liquid_content; idempotent reagent attachment by Bioyond materialId. - Phase 4: introduce SirnaResourceSynchronizer over BioyondResourceSynchronizer; install once in post_init without double-sync. - Phase 5: numeric stack orientation and bioyond_axis support in bioyond warehouses/__init__/decks (carries forward prior in-progress edits). - _resolve_location_to_warehouse now raises on ambiguity; deck constructor accepts warehouse_bioyond_ids kwarg.	2026-05-12 19:35:39 +08:00
yxz321	1f93740580	fix: add assert for convenient API call only when DEBUG_CLI_ENABLED	2026-05-12 19:35:39 +08:00
yxz321	98c27cde40	feat: RNA add guided siRNA manual load gate - Expose siRNA order and material handles for manual-confirm load workflows. - Gate scheduler start on explicit material-load confirmation before calling Bioyond RPC. - Improve lazy API config diagnostics and Sirna warehouse/material resource handling.	2026-05-12 19:35:39 +08:00
yxz321	18c3263e92	feat: RNA refine Bioyond siRNA Experiment 1 submission	2026-05-12 19:35:39 +08:00
yxz321	1519a7d985	feat: RNA add Bioyond siRNA station resources and Experiment 1 submission - Add siRNA station runtime, decorator metadata, and lazy init - Implement Experiment 1 submit, start, and reset flows - Add siRNA deck and numeric warehouse stack resources - Move siRNA example config to temp_benyao	2026-05-12 19:35:39 +08:00
yxz321	96c3f5a3e5	feat: RNA. Initial sirna workstation implementation.	2026-05-12 19:35:39 +08:00
		`@@ -0,0 +1,3 @@`
							`from .sirna_station import BioyondSirnaStation, fetch_workflow_list, load_sirna_config`

							`__all__ = ["BioyondSirnaStation", "fetch_workflow_list", "load_sirna_config"]`
		`@@ -0,0 +1 @@`
							`from . import sirna_materials # noqa: F401 ensure @resource classes are importable for PLR deserialize`