agents: let trigger rotate_secret revoke the outgoing webhook secret (MARSOHS-1078)

src/pydo/agents/custom_triggers.py

@@ -157,16 +157,25 @@ def delete(self, trigger_id: str) -> None:
         """
         self._send("DELETE", f"{_TRIGGERS_PATH}/{_quote(trigger_id)}")
 
-    def rotate_secret(self, trigger_id: str) -> Any:
+    def rotate_secret(self, trigger_id: str, *, revoke_previous: bool = False) -> Any:
         """Issue a new webhook secret (``POST .../{id}/rotate-secret``).
 
-        Webhook triggers only (``409`` for cron). The new secret is shown
-        once; the previous value stays valid briefly for in-flight deliveries.
+        Webhook triggers only (``409`` for cron). The new secret is shown once.
+
+        By default the outgoing secret keeps verifying deliveries for a short
+        server-configured window, because the provider signs with the old value
+        until someone pastes the new one in, and ``previous_secret_expires_at``
+        in the response says when it dies. Pass ``revoke_previous=True`` to
+        retire it on this call instead — intended for a compromised secret,
+        since deliveries still signed with the old value fail immediately, and
+        the response then carries ``previous_secret_revoked`` instead of an
+        expiry. Exactly one of the two is present.
         """
         return self._parse_json(
             self._send(
                 "POST",
                 f"{_TRIGGERS_PATH}/{_quote(trigger_id)}/rotate-secret",
+                params={"revoke_previous": "true" if revoke_previous else None},
             ),
         )
 

src/pydo/aio/agents/custom_triggers.py

@@ -120,12 +120,27 @@ async def delete(self, trigger_id: str) -> None:
         """Soft-delete a trigger (``DELETE /v2/agents/triggers/{id}``)."""
         await self._send("DELETE", f"{_TRIGGERS_PATH}/{_quote(trigger_id)}")
 
-    async def rotate_secret(self, trigger_id: str) -> Any:
-        """Issue a new webhook secret (shown once)."""
+    async def rotate_secret(
+        self, trigger_id: str, *, revoke_previous: bool = False
+    ) -> Any:
+        """Issue a new webhook secret (``POST .../{id}/rotate-secret``).
+
+        Webhook triggers only (``409`` for cron). The new secret is shown once.
+
+        By default the outgoing secret keeps verifying deliveries for a short
+        server-configured window, because the provider signs with the old value
+        until someone pastes the new one in, and ``previous_secret_expires_at``
+        in the response says when it dies. Pass ``revoke_previous=True`` to
+        retire it on this call instead — intended for a compromised secret,
+        since deliveries still signed with the old value fail immediately, and
+        the response then carries ``previous_secret_revoked`` instead of an
+        expiry. Exactly one of the two is present.
+        """
         return await self._parse_json(
             await self._send(
                 "POST",
                 f"{_TRIGGERS_PATH}/{_quote(trigger_id)}/rotate-secret",
+                params={"revoke_previous": "true" if revoke_previous else None},
             ),
         )
 

tests/agents/test_async_triggers.py

@@ -59,6 +59,19 @@ def _make_async_resources(responses: List[_FakeAsyncResponse]) -> AsyncAgentsRes
     )
 
 
+def _find_call(resources: AsyncAgentsResources, path_suffix: str):
+    """Return the single recorded call whose URL path ends with path_suffix."""
+    matches = [
+        c
+        for c in resources._proxy._original._pipeline.calls
+        if c.request.url.split("?")[0].endswith(path_suffix)
+    ]
+    assert (
+        len(matches) == 1
+    ), f"expected exactly one {path_suffix} call, got {len(matches)}"
+    return matches[0]
+
+
 @pytest.mark.asyncio
 async def test_async_list_and_create_triggers():
     resources = _make_async_resources(
@@ -130,6 +143,11 @@ async def test_async_update_delete_rotate_and_executions():
 
     rotated = await resources.triggers.rotate_secret("t1")
     assert rotated.webhook_secret == "new"
+    # Located by URL rather than by index: this test walks a fixed sequence of
+    # calls, and a positional assertion silently starts checking someone else's
+    # request the moment a call is inserted above.
+    rotate_call = _find_call(resources, "/rotate-secret")
+    assert "revoke_previous" not in rotate_call.request.url
 
     executions = await resources.triggers.list_executions("t1")
     assert executions.executions[0].execution_id == "e1"
@@ -152,3 +170,21 @@ async def test_async_update_delete_rotate_and_executions():
     assert resources._proxy._original._pipeline.calls[7].request.url.endswith(
         "/v2/agents/webhook-providers"
     )
+
+
+@pytest.mark.asyncio
+async def test_async_rotate_secret_revoke_previous():
+    resources = _make_async_resources(
+        [
+            _FakeAsyncResponse(
+                200, {"webhook_secret": "new", "previous_secret_revoked": True}
+            )
+        ]
+    )
+
+    rotated = await resources.triggers.rotate_secret("t1", revoke_previous=True)
+
+    call = resources._proxy._original._pipeline.calls[0]
+    assert call.request.method == "POST"
+    assert "revoke_previous=true" in call.request.url
+    assert rotated.previous_secret_revoked is True

tests/agents/test_triggers.py

@@ -197,15 +197,33 @@ def test_delete_trigger_returns_none_on_204():
 
 
 def test_rotate_secret():
-    body = {"webhook_secret": "whsec_rotated"}
+    body = {
+        "webhook_secret": "whsec_rotated",
+        "previous_secret_expires_at": "2026-07-01T12:05:00Z",
+    }
     resources = _make_resources([_FakeResponse(200, body)])
 
     resp = resources.triggers.rotate_secret("t1")
 
     call = _last_call(resources)
     assert call.request.method == "POST"
     assert call.request.url.endswith("/v2/agents/triggers/t1/rotate-secret")
+    assert "revoke_previous" not in call.request.url
     assert resp.webhook_secret == "whsec_rotated"
+    assert resp.previous_secret_expires_at == "2026-07-01T12:05:00Z"
+
+
+def test_rotate_secret_revoke_previous():
+    body = {"webhook_secret": "whsec_rotated", "previous_secret_revoked": True}
+    resources = _make_resources([_FakeResponse(200, body)])
+
+    resp = resources.triggers.rotate_secret("t1", revoke_previous=True)
+
+    call = _last_call(resources)
+    assert call.request.method == "POST"
+    assert _path(call.request.url).endswith("/v2/agents/triggers/t1/rotate-secret")
+    assert "revoke_previous=true" in call.request.url
+    assert resp.previous_secret_revoked is True
 
 
 # ---------------------------------------------------------------------------