Z-slice segment-counting line (negative result)

benchmark/testset.py

@@ -41,6 +41,8 @@ class TestCase:
     image_b64: str  # Combined view (for backward compatibility)
     top_image_b64: Optional[str]  # TOP view only
     side_image_b64: Optional[str]  # SIDE view only
+    midplane_b64: Optional[str]  # Single XY slice at z=Z//2 (no projection)
+    zslices_b64: Optional[list[str]]  # XY slices at _ZSLICE_FRACTIONS of Z
     volume: Optional[np.ndarray]
     ground_truth_stage: Optional[str]
 
@@ -215,6 +217,57 @@ def _create_three_view_image(volume: np.ndarray, max_dim: int = 1500) -> str:
     return base64.b64encode(buffer.getvalue()).decode("utf-8")
 
 
+def _create_slice_image(
+    volume: np.ndarray, z: Optional[int] = None, max_dim: int = 800
+) -> str:
+    """Render a single XY z-slice as base64 JPEG.
+
+    Unlike the projection helpers, this preserves depth information at one
+    plane: overlapping body segments stay separated rather than fusing.
+    Uses the same crop as the three-view projection so framing matches.
+
+    Parameters
+    ----------
+    volume : np.ndarray
+        3D volume array (Z, Y, X)
+    z : int, optional
+        Slice index. Defaults to the midplane (Z // 2).
+    max_dim : int
+        Maximum output dimension in pixels.
+    """
+    _ensure_dependencies()
+
+    if z is None:
+        z = volume.shape[0] // 2
+    z = max(0, min(z, volume.shape[0] - 1))
+
+    bounds = _compute_crop_bounds(volume)
+    slice_img = volume[z, bounds[0]:bounds[1], bounds[2]:bounds[3]]
+    slice_norm = _normalize_image(slice_img)
+
+    pil_img = PIL_Image.fromarray(slice_norm)
+    if max(pil_img.size) > max_dim:
+        scale = max_dim / max(pil_img.size)
+        new_size = (int(pil_img.size[0] * scale), int(pil_img.size[1] * scale))
+        pil_img = pil_img.resize(new_size, PIL_Image.Resampling.LANCZOS)
+
+    buffer = io.BytesIO()
+    pil_img.save(buffer, format="JPEG", quality=90)
+    return base64.b64encode(buffer.getvalue()).decode("utf-8")
+
+
+_ZSLICE_FRACTIONS = (0.25, 0.40, 0.50, 0.60, 0.75)
+
+
+def _create_zslice_stack(volume: np.ndarray, max_dim: int = 800) -> list[str]:
+    """Render XY slices at fixed Z fractions as a list of base64 JPEGs."""
+    z_max = volume.shape[0]
+    return [
+        _create_slice_image(volume, z=int(round(f * (z_max - 1))), max_dim=max_dim)
+        for f in _ZSLICE_FRACTIONS
+    ]
+
+
 def _create_separate_view_images(volume: np.ndarray, max_dim: int = 1000) -> Tuple[str, str]:
     """Create separate TOP and SIDE view images from volume, return base64 tuple.
 
@@ -359,15 +412,13 @@ def iter_embryo(
             volume = _load_volume(vol_path) if self.load_volumes else None
 
             # Create images
-            if volume is not None:
-                image_b64 = _create_three_view_image(volume)
-                top_b64, side_b64 = _create_separate_view_images(volume)
-            else:
-                # Load just for image if not loading full volumes
-                temp_vol = _load_volume(vol_path)
-                image_b64 = _create_three_view_image(temp_vol)
-                top_b64, side_b64 = _create_separate_view_images(temp_vol)
-                del temp_vol
+            vol = volume if volume is not None else _load_volume(vol_path)
+            image_b64 = _create_three_view_image(vol)
+            top_b64, side_b64 = _create_separate_view_images(vol)
+            midplane_b64 = _create_slice_image(vol)
+            zslices_b64 = _create_zslice_stack(vol)
+            if volume is None:
+                del vol
 
             # Get ground truth
             gt_stage = self.ground_truth.get_stage_at(embryo_id, timepoint)
@@ -378,6 +429,8 @@ def iter_embryo(
                 image_b64=image_b64,
                 top_image_b64=top_b64,
                 side_image_b64=side_b64,
+                midplane_b64=midplane_b64,
+                zslices_b64=zslices_b64,
                 volume=volume,
                 ground_truth_stage=gt_stage,
             )

experiments/program.md

@@ -52,9 +52,13 @@ async def perceive_xxx(
     references: dict[str, list[str]],  # stage -> [base64 reference images]
     history: list[dict],         # previous timepoints: [{timepoint, stage}]
     timepoint: int,              # current timepoint number
+    midplane_b64: str | None = None,  # optional: single XY z-slice at z=Z//2
 ) -> PerceptionOutput:           # from perception._base
 ```
 
+`midplane_b64` is only passed if your function declares it (or `**kwargs`).
+The runner inspects the signature, so existing variants don't need to change.
+
 ## Available Utilities (from `perception._base`)
 
 - `call_claude(system, content)` — single API call, returns response text

perception/__init__.py

@@ -41,6 +41,9 @@ def get_functions() -> dict:
     from .compare import perceive_compare
     from .changegate import perceive_changegate
     from .duration_aware import perceive_duration_aware
+    from .zslice import perceive_zslice
+    from .zslice_asym import perceive_zslice_asym
+    from .zslice_multi import perceive_zslice_multi
 
     _FUNCTIONS = {
         "minimal": perceive_minimal,
@@ -58,5 +61,8 @@ def get_functions() -> dict:
         "compare": perceive_compare,
         "changegate": perceive_changegate,
         "duration_aware": perceive_duration_aware,
+        "zslice": perceive_zslice,
+        "zslice_asym": perceive_zslice_asym,
+        "zslice_multi": perceive_zslice_multi,
     }
     return _FUNCTIONS

perception/zslice.py

@@ -0,0 +1,118 @@
+"""
+Z-slice subagent perception.
+
+Max-intensity projections collapse depth: overlapping body segments fuse into
+one bright band, destroying the discriminative feature for fold stages — how
+many times the body folds back on itself. A biologist resolves this by
+scrolling z-slices and counting body-segment cross-sections at the midplane.
+
+This variant runs the scientific classifier on the projection, then for
+fold-region timepoints sends the midplane z-slice to a subagent that counts
+discrete body-segment profiles. The count is mapped to a stage and overrides
+the primary call.
+"""
+
+import re
+
+from ._base import PerceptionOutput, call_claude, parse_stage_json
+from .scientific import perceive_scientific
+
+FOLD_STAGES = {"1.5fold", "2fold", "pretzel"}
+
+SEGMENT_SYSTEM = """\
+You are analyzing a single optical z-slice through a folded C. elegans embryo \
+inside its eggshell. This is NOT a max-intensity projection — it is one plane, \
+so body segments that overlap in projections appear as separate cross-sections \
+here.
+
+Count the number of DISCRETE body-segment cross-sections visible inside the \
+eggshell. A segment cross-section is a bright, roughly elliptical or band-like \
+region of fluorescent nuclei. Segments separated by a clear dark gap are \
+distinct; segments touching or merged count as one.
+
+Typical counts by stage:
+- 1 segment  → comma or early 1.5fold (body not yet folded back at this plane)
+- 2 segments → 2fold (one hairpin: two parallel passes)
+- 3+ segments → pretzel (multiple coils crossing this plane)
+
+Respond with JSON:
+{
+  "segment_count": <integer>,
+  "reasoning": "Brief description of what you counted"
+}"""
+
+
+def _segment_count_to_stage(count: int) -> str:
+    if count >= 3:
+        return "pretzel"
+    if count == 2:
+        return "2fold"
+    return "1.5fold"
+
+
+def _parse_segment_count(raw: str) -> int | None:
+    data = parse_stage_json(raw)
+    val = data.get("segment_count")
+    if isinstance(val, int):
+        return val
+    if isinstance(val, str):
+        m = re.search(r"\d+", val)
+        if m:
+            return int(m.group(0))
+    return None
+
+
+async def _count_segments(midplane_b64: str) -> tuple[int | None, str]:
+    """Ask the subagent to count body-segment cross-sections in the midplane slice."""
+    content = [
+        {
+            "type": "image",
+            "source": {"type": "base64", "media_type": "image/jpeg", "data": midplane_b64},
+        },
+        {
+            "type": "text",
+            "text": (
+                "Count the discrete body-segment cross-sections visible in this "
+                "single z-slice. Respond with the JSON format above."
+            ),
+        },
+    ]
+    raw = await call_claude(system=SEGMENT_SYSTEM, content=content, max_tokens=512)
+    return _parse_segment_count(raw), raw
+
+
+async def perceive_zslice(
+    image_b64: str,
+    references: dict[str, list[str]],
+    history: list[dict],
+    timepoint: int,
+    midplane_b64: str | None = None,
+) -> PerceptionOutput:
+    """Scientific classifier with a z-slice segment-counting subagent on fold stages."""
+    primary = await perceive_scientific(image_b64, references, history, timepoint)
+
+    if midplane_b64 is None or primary.stage not in FOLD_STAGES:
+        return primary
+
+    count, sub_raw = await _count_segments(midplane_b64)
+    if count is None:
+        return PerceptionOutput(
+            stage=primary.stage,
+            reasoning=f"{primary.reasoning} | [zslice] parse failed; kept primary",
+            verification_triggered=True,
+            phase_count=2,
+            raw_response=primary.raw_response,
+        )
+
+    sub_stage = _segment_count_to_stage(count)
+    return PerceptionOutput(
+        stage=sub_stage,
+        reasoning=(
+            f"{primary.reasoning} | [zslice] {count} segment(s) at midplane → "
+            f"{sub_stage}"
+            + ("" if sub_stage == primary.stage else f" (overrode {primary.stage})")
+        ),
+        verification_triggered=True,
+        phase_count=2,
+        raw_response=sub_raw,
+    )

perception/zslice_asym.py

@@ -0,0 +1,59 @@
+"""
+Z-slice subagent with asymmetric (upgrade-only) override.
+
+The original zslice variant regressed pretzel 88% -> 7% because a single
+midplane slice rarely shows >=3 separable cross-sections on a pretzel embryo
+(body folds sit at different z-depths). This variant keeps the segment-counting
+subagent but only lets it move the prediction *forward* in the fold ordering
+(1.5fold -> 2fold -> pretzel), never back.
+
+Trigger excludes pretzel since there is nothing to upgrade to, so the worst
+case on that stage is parity with `scientific`.
+"""
+
+from ._base import PerceptionOutput
+from .scientific import perceive_scientific
+from .zslice import _count_segments, _segment_count_to_stage
+
+_FOLD_ORDER = {"1.5fold": 0, "2fold": 1, "pretzel": 2}
+_TRIGGER_STAGES = {"1.5fold", "2fold"}
+
+
+async def perceive_zslice_asym(
+    image_b64: str,
+    references: dict[str, list[str]],
+    history: list[dict],
+    timepoint: int,
+    midplane_b64: str | None = None,
+) -> PerceptionOutput:
+    """Scientific classifier + upgrade-only z-slice override on 1.5fold/2fold."""
+    primary = await perceive_scientific(image_b64, references, history, timepoint)
+
+    if midplane_b64 is None or primary.stage not in _TRIGGER_STAGES:
+        return primary
+
+    count, sub_raw = await _count_segments(midplane_b64)
+    if count is None:
+        return PerceptionOutput(
+            stage=primary.stage,
+            reasoning=f"{primary.reasoning} | [zslice-asym] parse failed; kept primary",
+            verification_triggered=True,
+            phase_count=2,
+            raw_response=primary.raw_response,
+        )
+
+    sub_stage = _segment_count_to_stage(count)
+    if _FOLD_ORDER[sub_stage] > _FOLD_ORDER[primary.stage]:
+        final = sub_stage
+        note = f"{count} segment(s) -> {sub_stage} (upgraded from {primary.stage})"
+    else:
+        final = primary.stage
+        note = f"{count} segment(s) -> {sub_stage}; not above primary, kept {primary.stage}"
+
+    return PerceptionOutput(
+        stage=final,
+        reasoning=f"{primary.reasoning} | [zslice-asym] {note}",
+        verification_triggered=True,
+        phase_count=2,
+        raw_response=sub_raw,
+    )

perception/zslice_multi.py

@@ -0,0 +1,67 @@
+"""
+Multi-plane z-slice subagent: max segment-count across several depths.
+
+The single-midplane variants (zslice, zslice_asym) failed because one z-plane
+rarely intersects >=3 body segments at separable locations -- folds sit at
+different depths. This variant counts segments at five z-fractions
+(0.25/0.40/0.50/0.60/0.75 of Z) and takes the *maximum*. The hypothesis is
+that a pretzel embryo will show >=3 segments on at least one plane even if no
+single plane does.
+
+Override is symmetric (full FOLD_STAGES trigger) so the hypothesis is tested
+cleanly; if max-count is still <=2 on pretzel, asymmetric damage control can
+be layered on afterward.
+"""
+
+import asyncio
+
+from ._base import PerceptionOutput
+from .scientific import perceive_scientific
+from .zslice import FOLD_STAGES, _count_segments, _segment_count_to_stage
+
+
+async def _max_segment_count(zslices_b64: list[str]) -> tuple[int | None, list[int | None]]:
+    """Count segments on each slice concurrently; return (max, per-slice list)."""
+    results = await asyncio.gather(*(_count_segments(b64) for b64 in zslices_b64))
+    counts: list[int | None] = [c for c, _ in results]
+    valid = [c for c in counts if c is not None]
+    return (max(valid) if valid else None), counts
+
+
+async def perceive_zslice_multi(
+    image_b64: str,
+    references: dict[str, list[str]],
+    history: list[dict],
+    timepoint: int,
+    zslices_b64: list[str] | None = None,
+) -> PerceptionOutput:
+    """Scientific classifier + max-count override across multiple z-slices."""
+    primary = await perceive_scientific(image_b64, references, history, timepoint)
+
+    if not zslices_b64 or primary.stage not in FOLD_STAGES:
+        return primary
+
+    max_count, per_slice = await _max_segment_count(zslices_b64)
+    counts_str = "/".join("-" if c is None else str(c) for c in per_slice)
+
+    if max_count is None:
+        return PerceptionOutput(
+            stage=primary.stage,
+            reasoning=f"{primary.reasoning} | [zslice-multi] all parses failed; kept primary",
+            verification_triggered=True,
+            phase_count=2,
+            raw_response=primary.raw_response,
+        )
+
+    sub_stage = _segment_count_to_stage(max_count)
+    return PerceptionOutput(
+        stage=sub_stage,
+        reasoning=(
+            f"{primary.reasoning} | [zslice-multi] counts {counts_str} max={max_count} -> "
+            f"{sub_stage}"
+            + ("" if sub_stage == primary.stage else f" (overrode {primary.stage})")
+        ),
+        verification_triggered=True,
+        phase_count=2,
+        raw_response=primary.raw_response,
+    )

program.md

@@ -52,9 +52,13 @@ async def perceive_xxx(
     references: dict[str, list[str]],  # stage -> [base64 reference images]
     history: list[dict],         # previous timepoints: [{timepoint, stage}]
     timepoint: int,              # current timepoint number
+    midplane_b64: str | None = None,  # optional: single XY z-slice at z=Z//2
 ) -> PerceptionOutput:           # from perception._base
 ```
 
+`midplane_b64` is only passed if your function declares it (or `**kwargs`).
+The runner inspects the signature, so existing variants don't need to change.
+
 ## Available Utilities (from `perception._base`)
 
 - `call_claude(system, content)` — single API call, returns response text