canary: detect provider API schema drift; release v0.7.0

The export reads ChatGPT's and Claude's undocumented internal web APIs,
which can change shape without notice; the worst failure for a backup tool
is a silent one (skipped/mis-parsed content with no error). Add a `canary`
command + BaseProvider.check_drift() (overridden by ChatGPT/Claude) that
fetches one listing page + one conversation per provider and asserts only
the normalizer's load-bearing fields — not the full response shape, which
churns harmlessly. The top silent risk it guards is a renamed retrieval-tool
author bypassing the hidden-content collapse. ERROR findings exit non-zero;
WARN findings are surfaced but non-fatal so a backup run is never blocked.

Also retires the in-app watch mode and headless StartOS direction from the
roadmap (tool stays a local, manually-run CLI) and updates docs.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
JesseMarkowitz
2026-06-28 01:37:24 -04:00
co-authored by Claude Opus 4.8
parent ef603cf659
commit 1e016ea652
10 changed files with 693 additions and 80 deletions
+143
View File
@@ -973,3 +973,146 @@ class TestChatGPTSessionHealth:
ok, detail = p.session_health()
assert ok is False
assert "unreachable" in detail.lower()
# ---------------------------------------------------------------------------
# API-drift canary (§10): assert load-bearing fields, flag silent risks.
# ---------------------------------------------------------------------------
def _sev(findings):
return {f["severity"] for f in findings}
def _checks(findings, severity):
return {f["check"] for f in findings if f["severity"] == severity}
class TestChatGPTDriftCanary:
def _provider(self, page, detail):
from src.providers.chatgpt import ChatGPTProvider
p = ChatGPTProvider.__new__(ChatGPTProvider)
p.list_conversations = lambda offset=0, limit=5: page
p.get_conversation = lambda cid: detail
return p
@staticmethod
def _detail(mapping):
return {
"conversation_id": "c1", "title": "T",
"create_time": 1.0, "update_time": 2.0, "mapping": mapping,
}
@staticmethod
def _node(role="user", name=None, content_type="text", parts=("hello",)):
return {
"id": "n", "parent": None, "children": [],
"message": {
"author": {"role": role, "name": name},
"content": {"content_type": content_type, "parts": list(parts)},
"metadata": {},
},
}
def test_healthy_shape_is_ok_only(self):
from src.providers.base import DRIFT_OK
p = self._provider([{"id": "c1", "title": "T", "update_time": "x"}],
self._detail({"n1": self._node()}))
assert _sev(p.check_drift()) == {DRIFT_OK}
def test_missing_listing_id_is_error(self):
from src.providers.base import DRIFT_ERROR
p = self._provider([{"title": "T"}], self._detail({"n1": self._node()}))
assert DRIFT_ERROR in _sev(p.check_drift())
def test_empty_mapping_is_error(self):
from src.providers.base import DRIFT_ERROR
p = self._provider([{"id": "c1", "title": "T", "update_time": "x"}],
self._detail({}))
findings = p.check_drift()
assert "detail" in _checks(findings, DRIFT_ERROR)
def test_unfamiliar_tool_author_warns_collapse(self):
from src.providers.base import DRIFT_WARN
mapping = {
"n1": self._node(), # a healthy text msg so parts check passes
"n2": self._node(role="tool", name="brand_new_retriever"),
}
p = self._provider([{"id": "c1", "title": "T", "update_time": "x"}],
self._detail(mapping))
assert "collapse" in _checks(p.check_drift(), DRIFT_WARN)
def test_new_content_type_warns(self):
from src.providers.base import DRIFT_WARN
mapping = {
"n1": self._node(),
"n2": self._node(content_type="hologram_v2", parts=()),
}
p = self._provider([{"id": "c1", "title": "T", "update_time": "x"}],
self._detail(mapping))
assert "content_type" in _checks(p.check_drift(), DRIFT_WARN)
def test_known_tool_author_not_flagged(self):
# file_search is a known collapse author — must NOT warn.
mapping = {
"n1": self._node(),
"n2": self._node(role="tool", name="file_search"),
}
p = self._provider([{"id": "c1", "title": "T", "update_time": "x"}],
self._detail(mapping))
assert "collapse" not in _checks(p.check_drift(), "warn")
class TestClaudeDriftCanary:
def _provider(self, page, detail):
from src.providers.claude import ClaudeProvider
p = ClaudeProvider.__new__(ClaudeProvider)
p.list_conversations = lambda offset=0, limit=5: page
p.get_conversation = lambda cid: detail
return p
@staticmethod
def _detail(msgs):
return {"uuid": "u1", "name": "N", "created_at": "x", "updated_at": "y",
"chat_messages": msgs}
def test_healthy_flat_shape_is_ok(self):
from src.providers.base import DRIFT_OK
msgs = [{"uuid": "m1", "sender": "human", "text": "hi",
"created_at": "x", "attachments": [], "files": []}]
p = self._provider([{"uuid": "u1", "name": "N", "updated_at": "z"}],
self._detail(msgs))
assert _sev(p.check_drift()) == {DRIFT_OK}
def test_content_as_list_warns(self):
from src.providers.base import DRIFT_WARN
msgs = [{"uuid": "m1", "sender": "assistant",
"content": [{"type": "text", "text": "hi"}], "created_at": "x"}]
p = self._provider([{"uuid": "u1", "name": "N", "updated_at": "z"}],
self._detail(msgs))
assert "content" in _checks(p.check_drift(), DRIFT_WARN)
def test_nonempty_attachments_warn(self):
from src.providers.base import DRIFT_WARN
msgs = [{"uuid": "m1", "sender": "human", "text": "hi",
"created_at": "x", "attachments": [{"file_name": "a.pdf"}]}]
p = self._provider([{"uuid": "u1", "name": "N", "updated_at": "z"}],
self._detail(msgs))
assert "attachments" in _checks(p.check_drift(), DRIFT_WARN)
def test_empty_messages_is_error(self):
from src.providers.base import DRIFT_ERROR
p = self._provider([{"uuid": "u1", "name": "N", "updated_at": "z"}],
self._detail([]))
assert DRIFT_ERROR in _sev(p.check_drift())