ENG-696: structured publish result contract

cowork/harnesses/anton_harness/tools.py

@@ -111,7 +111,7 @@ def _access_from_state(entry: dict) -> dict:
 )
 
 
-async def _cowork_publish_or_preview(session: Any, tc_input: dict) -> str:
+async def _cowork_publish_or_preview(session: Any, tc_input: dict):
     """Server-side equivalent of anton.tools.handle_publish_or_preview.
 
     Mirrors the same `action` semantics:
@@ -200,6 +200,28 @@ async def _cowork_publish_or_preview(session: Any, tc_input: dict) -> str:
     # action == 'publish' — delegate to the service (single source of truth for
     # target resolution, fullstack bundling, vault secrets, access, history,
     # and report_id reuse).
+    #
+    # The SideEffectResult envelope (ENG-696) lives in anton; cowork-server and
+    # anton deploy independently, so an older anton without it must not break
+    # publishing. When absent, fall back to the plain `message` string — the
+    # exact pre-envelope return value, so behavior is byte-identical there.
+    try:
+        from anton.core.tools.side_effect import SideEffectResult, now_iso
+    except ImportError:
+        SideEffectResult = None
+
+    def _ok(message: str, **fields):
+        if SideEffectResult is None:
+            return message
+        return SideEffectResult(
+            success=True, message=message, committed_at=now_iso(), **fields
+        ).to_outcome()
+
+    def _fail(message: str, reason: str = ""):
+        if SideEffectResult is None:
+            return message
+        return SideEffectResult.failed(message, reason=reason)
+
     try:
         result = _publish_artifact(abs_path, access=access)
     except ValueError as exc:
@@ -210,21 +232,38 @@ async def _cowork_publish_or_preview(session: Any, tc_input: dict) -> str:
         # they already have and block a legitimate retry.
         if "api key" in str(exc).lower():
             logger.info("Cowork publish blocked: %s", exc)
-            return (
+            return _fail(
                 "STOP: No Minds API key configured. Tell the user to set their "
                 "Minds API key in Settings (or in their .env) before publishing. "
-                "Do NOT call this tool again until they confirm the key is set."
+                "Do NOT call this tool again until they confirm the key is set.",
+                reason="missing_api_key",
             )
         logger.info("Cowork publish rejected: %s", exc)
-        return f"PUBLISH FAILED: {exc}"
+        return _fail(f"PUBLISH FAILED: {exc}", reason="ValueError")
     except Exception as exc:
         logger.exception("Cowork publish tool failed")
-        return f"PUBLISH FAILED: {exc}"
+        return _fail(f"PUBLISH FAILED: {exc}", reason=type(exc).__name__)
 
+    inner = result.get("result", {}) if isinstance(result, dict) else {}
     view_url = result.get("url", "") if isinstance(result, dict) else ""
+    report_id = inner.get("report_id") or None
+    md5 = inner.get("md5") or None
     if not view_url:
-        return "Published, but no view URL was returned."
-    return f"Published successfully! View URL: {view_url}"
+        # Committed (the service returned) but the URL is missing — success is
+        # true, external_url stays null so the ambiguity is explicit, not prose.
+        return _ok(
+            "Published, but no view URL was returned.",
+            resource_id=report_id,
+            idempotency_key=report_id,
+            content_hash=(f"md5:{md5}" if md5 else None),
+        )
+    return _ok(
+        f"Published successfully! View URL: {view_url}",
+        resource_id=report_id,
+        external_url=view_url,
+        idempotency_key=report_id,
+        content_hash=(f"md5:{md5}" if md5 else None),
+    )
 
 
 def build_cowork_publish_tool():

tests/test_harness_publish_access.py

@@ -35,7 +35,7 @@ async def test_harness_forwards_explicit_password(tmp_path):
         )
     _, kwargs = fake.call_args
     assert kwargs["access"] == {"mode": "password", "password": "hunter2"}
-    assert "Published" in out
+    assert "Published" in getattr(out, "content", out)
 
 
 @pytest.mark.asyncio

tests/test_stable_publish_url.py

@@ -253,10 +253,28 @@ def _fake_publish_artifact(raw_path, access=None):
             _FakeSession(tmp_path),
             {"file_path": str(root / "static" / "index.html"), "action": "publish", "title": "Dash"},
         ))
-    assert "https://4nton.ai/a/uuid-1" in out
+    assert "https://4nton.ai/a/uuid-1" in getattr(out, "content", out)
     assert captured["path"].endswith("static/index.html")
 
 
+def test_tool_publish_falls_back_to_string_on_old_anton(tmp_path: Path):
+    # An anton without the SideEffectResult envelope (ENG-696) must still
+    # publish — the tool returns the plain pre-envelope string. Setting the
+    # module to None in sys.modules makes the `from ... import` raise ImportError.
+    import sys
+
+    root = _make_fullstack(tmp_path)
+    with patch.object(tools_mod, "_publish_artifact",
+                      lambda p, access=None: {"status": "ok", "url": "https://4nton.ai/a/uuid-1"}), \
+         patch.dict(sys.modules, {"anton.core.tools.side_effect": None}):
+        out = _run(tools_mod._cowork_publish_or_preview(
+            _FakeSession(tmp_path),
+            {"file_path": str(root / "static" / "index.html"), "action": "publish", "title": "Dash"},
+        ))
+    assert isinstance(out, str)
+    assert out == "Published successfully! View URL: https://4nton.ai/a/uuid-1"
+
+
 def test_tool_publish_no_api_key_returns_stop(tmp_path: Path):
     root = _make_fullstack(tmp_path)
 
@@ -268,7 +286,7 @@ def _raise(raw_path, access=None):
             _FakeSession(tmp_path),
             {"file_path": str(root / "static" / "index.html"), "action": "publish", "title": "Dash"},
         ))
-    assert "STOP" in out and "API key" in out
+    assert "STOP" in getattr(out, "content", out) and "API key" in getattr(out, "content", out)
 
 
 def test_tool_publish_unsupported_type_is_not_treated_as_missing_key(tmp_path: Path):
@@ -283,9 +301,10 @@ def _raise(raw_path, access=None):
             _FakeSession(tmp_path),
             {"file_path": str(root / "static" / "index.html"), "action": "publish", "title": "Dash"},
         ))
-    assert "STOP" not in out
-    assert "PUBLISH FAILED" in out
-    assert "Only HTML and Markdown" in out
+    text = getattr(out, "content", out)
+    assert "STOP" not in text
+    assert "PUBLISH FAILED" in text
+    assert "Only HTML and Markdown" in text
 
 
 # ---------------------------------------------------------------------------