' in page.text
- assert page.text.index('id="memory-detail"') < page.text.index('id="memory-editor"')
- stylesheet = client.get("/v2-assets/ledger.css")
- assert ".library-detail-stack { display: grid; gap: 12px; align-content: start; }" in stylesheet.text
-
-
-def test_workspace_switcher_uses_the_active_ledger_theme_for_native_dropdowns(monkeypatch, tmp_path):
- with _client(monkeypatch, tmp_path) as client:
- stylesheet = client.get("/v2-assets/ledger.css")
- assert stylesheet.status_code == 200
- css = stylesheet.text
- assert ".workspace-switcher select {" in css
- assert "background: var(--c-inset);" in css
- assert "color-scheme: dark;" in css
- assert 'body[data-theme="paper"] .workspace-switcher select { color-scheme: light; }' in css
- assert ".workspace-switcher select option { background: var(--c-inset); color: var(--c-fg); }" in css
- assert ".workspace-switcher select option:checked { background: var(--c-acc); color: var(--c-bg); }" in css
-
-
-def test_sidebar_keeps_manage_and_compare_plans_in_separate_flex_rows(monkeypatch, tmp_path):
- with _client(monkeypatch, tmp_path) as client:
- stylesheet = client.get("/v2-assets/ledger.css")
- assert stylesheet.status_code == 200
- css = stylesheet.text
- sidebar = css[css.index(".sidebar {"):css.index(".brand-row {")]
- assert "display: flex;" in sidebar
- assert "flex-direction: column;" in sidebar
- assert "grid-template-rows" not in sidebar
- assert ".primary-nav { flex: 1 0 auto; }" in css
- assert ".manage-nav { flex: 0 0 auto; }" in css
- assert ".sidebar-promo {\n flex: 0 0 auto;" in css
-
-
-def test_dashboard_grounded_answer_route_cites_or_abstains(monkeypatch, tmp_path):
- with _client(monkeypatch, tmp_path) as client:
- grounded = client.post(
- "/api/answer",
- json={
- "query": "Which database is the main database?",
- "workspace": "demo",
- "k": 8,
- "max_citations": 5,
- "candidate_depth": "adaptive",
- },
- )
- assert grounded.status_code == 200
- body = grounded.json()
- assert body["query"] == "Which database is the main database?"
- assert body["grounded"] is True
- assert body["abstained"] is False
- assert body["citations"]
- assert body["sources"] == body["citations"]
- assert "[1]" in body["answer"]
- assert body["candidate_depth"] == "adaptive"
- # ``candidate_k_used`` is the final page depth after prompt-safe
- # overfetch/widening, rather than the adaptive policy's starting depth.
- assert body["candidate_k_used"] >= body["candidate_k_requested"]
-
- abstained = client.post(
- "/api/answer",
- json={
- "query": "How should I bake a sourdough loaf?",
- "workspace": "demo",
- },
- )
- assert abstained.status_code == 200
- assert abstained.json()["grounded"] is False
- assert abstained.json()["abstained"] is True
-
-
-def test_dashboard_grounded_answer_route_bounds_and_redacts(monkeypatch, tmp_path):
- with _client(monkeypatch, tmp_path) as client:
- assert client.post("/api/answer", json={"query": "", "workspace": "demo"}).status_code == 422
- assert client.post(
- "/api/answer",
- json={"query": "database", "workspace": "demo", "k": 51},
- ).status_code == 422
-
-
-def test_team_account_routes_are_not_in_public_runtime(monkeypatch, tmp_path):
- with _client(monkeypatch, tmp_path) as client:
- assert client.post("/api/auth/setup", json={}).status_code == 404
- assert client.get("/api/auth/users").status_code == 404
- state = client.get("/api/auth/state").json()
- assert state["enabled"] is False
- assert state["hosted_team"] is True
-
-
-def test_local_agent_write_has_no_client_side_team_paywall(monkeypatch, tmp_path):
- with _client(monkeypatch, tmp_path) as client:
- response = client.post(
- "/api/remember",
- json={"workspace": "demo", "content": "Queues use at-least-once delivery."},
- )
- assert response.status_code == 200
-
-
-def test_http_memory_api_exposes_world_timed_agent_writes_immediately(monkeypatch, tmp_path):
- with _client(monkeypatch, tmp_path) as client:
- old = client.post(
- "/api/remember",
- json={
- "workspace": "demo",
- "content": "The API rate limit is 100 requests per minute.",
- "valid_from": 1_000.0,
- "subject_key": "api.rate_limit",
- "claim_kind": "configured_value",
- },
- ).json()
- new = client.post(
- "/api/intent/remember",
- json={
- "workspace": "demo",
- "text": "The API rate limit is 500 requests per minute.",
- "valid_from": 2_000.0,
- "subject_key": "api.rate_limit",
- "claim_kind": "configured_value",
- },
- ).json()
-
- before = client.get(
- "/api/recall",
- params={
- "workspace": "demo",
- "q": "What is the API rate limit?",
- "as_of": 1_500.0,
- },
- )
- after = client.post(
- "/api/answer",
- json={
- "workspace": "demo",
- "query": "What is the API rate limit?",
- "as_of": 2_500.0,
- "min_support": 0.0,
- },
- )
-
- assert before.status_code == 200
- assert [memory["id"] for memory in before.json()["memories"]] == [old["id"]]
- assert after.status_code == 200
- assert after.json()["sources"]
- service = client.app.state.service
- assert service.store.get_memory(old["id"]).valid_from == 1_000.0
- assert service.store.get_memory(new["id"]).valid_from == 2_000.0
- assert service.store.get_memory(old["id"]).provenance["review_state"] == "approved"
- assert service.store.get_memory(new["id"]).provenance["review_state"] == "approved"
-
-
-def test_keyword_recall_fallback_keeps_bitemporal_visibility(monkeypatch, tmp_path):
- """A semantic-backend failure must not leak current facts into historical views."""
- with _client(monkeypatch, tmp_path) as client:
- svc = v2_api.service()
- workspace_id = svc.store.get_or_create_workspace("demo")
- old = {"id": svc.engine.remember(
- "The fallback retention setting was ten days.", workspace_id=workspace_id,
- scope=Scope.WORKSPACE, valid_from=1_000.0, resolve_conflicts=False,
- )}
- new = {"id": svc.engine.remember(
- "The fallback retention setting was thirty days.", workspace_id=workspace_id,
- scope=Scope.WORKSPACE, valid_from=2_000.0, resolve_conflicts=False,
- )}
- # The writes happened during this test, but the fixture models facts learned
- # before the requested historical system-time anchors.
- svc.store.conn.execute(
- "UPDATE memories SET ingested_at=100 WHERE id=?", (old["id"],)
- )
- svc.store.conn.execute(
- "UPDATE memories SET ingested_at=200 WHERE id=?", (new["id"],)
- )
- svc.store.conn.execute(
- "UPDATE memories SET valid_to=2000, valid_to_recorded_at=200, "
- "subject_key='retention.days', claim_kind='configured_value' "
- "WHERE id=?",
- (old["id"],),
- )
- svc.store.conn.commit()
- old_before = v2_api._keyword_search(
- "demo", "fallback retention", valid_at=1_500.0, known_at=3_000.0
- )
- old_known = v2_api._keyword_search(
- "demo", "fallback retention", valid_at=1_500.0, known_at=50.0
- )
- current = v2_api._keyword_search(
- "demo", "fallback retention", valid_at=2_500.0, known_at=3_000.0
- )
- closure_unknown = v2_api._keyword_search(
- "demo", "fallback retention", valid_at=2_500.0, known_at=150.0
- )
-
- assert [memory["id"] for memory in old_before] == [old["id"]]
- assert old_known == []
- assert [memory["id"] for memory in current] == [new["id"]]
- assert [memory["id"] for memory in closure_unknown] == [old["id"]]
- assert closure_unknown[0]["valid_to_recorded_at"] == 200.0
- assert closure_unknown[0]["subject_key"] == "retention.days"
- assert closure_unknown[0]["claim_kind"] == "configured_value"
-
- def incompatible_embedder(*_args, **_kwargs):
- raise ValueError("shapes (256,) and (384,) not aligned")
-
- monkeypatch.setattr(svc, "recall", incompatible_embedder)
- fallback = client.get(
- "/api/recall",
- params={
- "workspace": "demo", "q": "fallback retention",
- "valid_at": 2_500.0, "known_at": 150.0,
- },
- )
- assert fallback.status_code == 200
- assert fallback.json()["mode"] == "keyword"
- assert [item["id"] for item in fallback.json()["memories"]] == [old["id"]]
-
- compact_fallback = client.get(
- "/api/recall",
- params={
- "workspace": "demo", "q": "fallback retention", "response_mode": "compact",
- "token_budget": 0,
- },
- )
- payload = compact_fallback.json()
- assert compact_fallback.status_code == 200
- assert payload["mode"] == "keyword"
- assert payload["response_mode"] == "compact"
- assert payload["usage"]["budget_tokens"] == 0
- assert payload["usage"]["context_tokens"] == 0
- assert payload["memories"] and "content" not in payload["memories"][0]
-
-
-def test_keyword_recall_fallback_excludes_untrusted_memories(monkeypatch, tmp_path):
- """A degraded HTTP recall must enforce the same prompt eligibility boundary."""
- with _client(monkeypatch, tmp_path) as client:
- svc = v2_api.service()
- workspace_id = svc.store.get_or_create_workspace("demo")
- trusted = {"id": svc.engine.remember(
- "Fallback visibility trusted candidate.",
- workspace_id=workspace_id, scope=Scope.WORKSPACE,
- )}
- untrusted = svc.remember(
- "Fallback visibility untrusted candidate.",
- workspace="demo",
- source="sync",
- trusted=False,
- )
-
- def incompatible_embedder(*_args, **_kwargs):
- raise ValueError("shapes (256,) and (384,) not aligned")
-
- monkeypatch.setattr(svc, "recall", incompatible_embedder)
- response = client.get(
- "/api/recall",
- params={"workspace": "demo", "q": "fallback visibility candidate", "k": 1},
- )
-
- payload = response.json()
- assert response.status_code == 200
- assert payload["mode"] == "keyword"
- assert [memory["id"] for memory in payload["memories"]] == [trusted["id"]]
- assert untrusted["id"] not in {memory["id"] for memory in payload["memories"]}
- assert "untrusted candidate" not in repr(payload)
-
-
-def test_http_memory_api_rejects_backdated_agent_claim_supersession(
- monkeypatch, tmp_path
-):
- with _client(monkeypatch, tmp_path) as client:
- original = client.post(
- "/api/remember",
- json={
- "workspace": "demo",
- "content": "The deployment window is Friday afternoon.",
- "valid_from": 2_000.0,
- },
- ).json()
- service = v2_api.service()
- count_before = len(service.store.list_memories(include_invalid=True))
- rejected = client.post(
- "/api/remember",
- json={
- "workspace": "demo",
- "content": "The deployment window is Thursday afternoon.",
- "valid_from": 1_000.0,
- },
- )
-
- assert rejected.status_code == 400
- assert service.store.get_memory(original["id"]).valid_to is None
- assert len(service.store.list_memories(include_invalid=True)) == count_before
-
-
-def test_manual_consolidation_stays_local_but_dreaming_is_cloud_only(
- monkeypatch, tmp_path
-):
- with _client(monkeypatch, tmp_path) as client:
- manual = client.post(
- "/api/consolidate",
- json={"workspace": "demo", "dry_run": True, "infer": False},
- )
- assert manual.status_code == 200
- dream = client.post(
- "/api/consolidate",
- json={"workspace": "demo", "dry_run": True, "infer": True},
- )
- assert dream.status_code == 501
- assert dream.json()["detail"]["cloud_only"] is True
-
-
-def test_analytics_route_delegates_to_managed_compute(monkeypatch, tmp_path):
- monkeypatch.setattr(
- "engraphis.cloud_features.run_managed_job",
- lambda service, workspace, kind: {
- "result": {
- "kind": kind,
- "generation": 4,
- "totals": {"live": 1},
- }
- },
- )
- with _client(monkeypatch, tmp_path) as client:
- response = client.get("/api/analytics?workspace=demo")
- assert response.status_code == 200
- assert response.json()["kind"] == "analytics"
- assert response.json()["generation"] == 4
-
-
-def test_unconnected_automation_returns_a_structured_auth_error(monkeypatch, tmp_path):
- for name in (
- "ENGRAPHIS_CLOUD_ACCESS_TOKEN",
- "ENGRAPHIS_CLOUD_ORGANIZATION_ID",
- "ENGRAPHIS_CLOUD_COMPUTE_URL",
- "ENGRAPHIS_CLOUD_REFRESH_CREDENTIAL",
- "ENGRAPHIS_CLOUD_CONTROL_URL",
- ):
- monkeypatch.delenv(name, raising=False)
- monkeypatch.setenv("ENGRAPHIS_STATE_DIR", str(tmp_path / "unconnected-state"))
-
- with _client(monkeypatch, tmp_path) as client:
- response = client.get("/api/automation?workspace=demo")
-
- assert response.status_code == 401
- # The copy is ``_public_session_error(401)``: fixed, status-keyed, and actionable. The
- # generic placeholder told an unconnected customer nothing they could act on.
- assert response.json()["detail"] == {
- "error": "Connect this installation to Engraphis Cloud to use hosted features.",
- "managed_cloud": True,
- "transient": False,
- "code": "cloud_unconfigured",
- }
-
-
-def test_hosted_automation_accepts_the_cloud_policy_field(monkeypatch, tmp_path):
- saved = {}
-
- class _Cloud:
- def upload_snapshot(self, workspace_id, snapshot):
- return {"generation": snapshot["generation"]}
-
- def get_policy(self, workspace_id):
- return {"enabled": False, "cadence_minutes": 1440, "dream_enabled": False}
-
- def save_policy(self, workspace_id, policy):
- saved.update(policy)
- return {"version": 2}
-
- monkeypatch.setattr(
- "engraphis.cloud_features.build_managed_snapshot",
- lambda service, workspace: ("ws_cloud", {"generation": 1}),
- )
- monkeypatch.setattr(
- "engraphis.cloud_features.CloudFeatureClient.from_environment",
- lambda workspace_id=None: _Cloud(),
- )
- with _client(monkeypatch, tmp_path) as client:
- response = client.post(
- "/api/automation",
- json={"enabled": True, "dream_enabled": True, "cadence_hours": 12},
- )
- assert response.status_code == 200
- assert response.json()["dream_enabled"] is True
- assert saved["dream_enabled"] is True
-
-
-def test_first_hosted_automation_view_bootstraps_the_recommended_policy(
- monkeypatch, tmp_path
-):
- """A connected Pro/Team workspace starts maintaining itself without a toggle."""
-
- uploaded = []
- saved = []
-
- class _Cloud:
- organization_id = "org_test"
-
- def get_policy(self, workspace_id):
- # Version zero is the private Cloud's documented no-policy sentinel.
- return {"enabled": False, "cadence_minutes": 1440, "version": 0}
-
- def upload_snapshot(self, workspace_id, snapshot):
- uploaded.append((workspace_id, snapshot))
- return {"generation": snapshot["generation"]}
-
- def save_policy(self, workspace_id, policy):
- saved.append((workspace_id, policy))
- return {"version": 1}
-
- def list_jobs(self, workspace_id, *, limit=10):
- return {"jobs": []}
-
- monkeypatch.setattr(
- "engraphis.cloud_features.build_managed_snapshot",
- lambda service, workspace: ("ws_cloud", {"generation": 7}),
- )
- monkeypatch.setattr(
- "engraphis.cloud_features.CloudFeatureClient.from_environment",
- lambda workspace_id=None: _Cloud(),
- )
- with _client(monkeypatch, tmp_path) as client:
- response = client.get("/api/automation")
-
- assert response.status_code == 200
- assert response.json()["enabled"] is True
- assert response.json()["dream"] is True
- assert uploaded == [("ws_cloud", {"generation": 7})]
- assert saved == [("ws_cloud", {
- "enabled": True,
- "cadence_minutes": 1440,
- "dream_enabled": True,
- "dream_min_new": 25,
- "dream_idle_minutes": 15,
- "infer": False,
- })]
-
-
-def test_first_automation_policy_retry_does_not_upload_the_snapshot_twice(
- monkeypatch, tmp_path
-):
- """A failed policy write resumes after the already successful private upload."""
-
- from engraphis.cloud_features import CloudFeatureError
-
- uploaded = []
- saved = []
- builds = []
-
- class _Cloud:
- organization_id = "org_test"
-
- def get_policy(self, workspace_id):
- return {"enabled": False, "cadence_minutes": 1440, "version": 0}
-
- def upload_snapshot(self, workspace_id, snapshot):
- uploaded.append((workspace_id, snapshot))
- return {"generation": snapshot["generation"]}
-
- def save_policy(self, workspace_id, policy):
- saved.append((workspace_id, policy))
- if len(saved) == 1:
- raise CloudFeatureError(
- "Engraphis Cloud is temporarily unavailable.",
- status=503,
- transient=True,
- )
- return {"version": 1}
-
- def list_jobs(self, workspace_id, *, limit=10):
- return {"jobs": []}
-
- def _snapshot(service, workspace):
- builds.append(workspace)
- return "ws_cloud", {"generation": 7}
-
- monkeypatch.setattr("engraphis.cloud_features.build_managed_snapshot", _snapshot)
- monkeypatch.setattr(
- "engraphis.cloud_features.CloudFeatureClient.from_environment",
- lambda workspace_id=None: _Cloud(),
- )
- with _client(monkeypatch, tmp_path) as client:
- first = client.get("/api/automation")
- second = client.get("/api/automation")
-
- assert first.status_code == 503
- assert second.status_code == 200
- assert len(builds) == 1
- assert uploaded == [("ws_cloud", {"generation": 7})]
- assert len(saved) == 2
-
-
-def test_concurrent_first_automation_views_upload_one_snapshot(monkeypatch, tmp_path):
- """Parallel dashboard reads serialize the sensitive first-bootstrap upload."""
-
- uploaded = []
- saved = []
- started = threading.Event()
- release_upload = threading.Event()
-
- class _Cloud:
- organization_id = "org_concurrent"
-
- def get_policy(self, workspace_id):
- return {"enabled": False, "cadence_minutes": 1440, "version": 0}
-
- def upload_snapshot(self, workspace_id, snapshot):
- uploaded.append((workspace_id, snapshot))
- started.set()
- assert release_upload.wait(timeout=5)
- return {"generation": snapshot["generation"]}
-
- def save_policy(self, workspace_id, policy):
- saved.append((workspace_id, policy))
- return {"version": 1}
-
- def list_jobs(self, workspace_id, *, limit=10):
- return {"jobs": []}
-
- monkeypatch.setattr(
- "engraphis.cloud_features.build_managed_snapshot",
- lambda service, workspace: ("ws_cloud", {"generation": 7}),
- )
- monkeypatch.setattr(
- "engraphis.cloud_features.CloudFeatureClient.from_environment",
- lambda workspace_id=None: _Cloud(),
- )
- with _client(monkeypatch, tmp_path):
- with ThreadPoolExecutor(max_workers=2) as pool:
- first = pool.submit(v2_api.automation_get)
- assert started.wait(timeout=5)
- second = pool.submit(v2_api.automation_get)
- release_upload.set()
- assert first.result(timeout=5)["enabled"] is True
- follower = second.result(timeout=5)
- assert follower["enabled"] is True
- assert follower["version"] == 1
-
- assert uploaded == [("ws_cloud", {"generation": 7})]
- assert len(saved) == 1
-
-
-def test_reading_or_disabling_automation_never_uploads_memory_content(
- monkeypatch, tmp_path
-):
- saved = {}
-
- class _Cloud:
- def get_policy(self, workspace_id):
- return {"enabled": True, "cadence_minutes": 60, "dream_enabled": True}
-
- def list_jobs(self, workspace_id, *, limit=10):
- return {"jobs": []}
-
- def save_policy(self, workspace_id, policy):
- saved.update(policy)
- return {"version": 3}
-
- def _unexpected_upload(*args, **kwargs):
- raise AssertionError("policy inspection must not build or upload a snapshot")
-
- monkeypatch.setattr(
- "engraphis.cloud_features.build_managed_snapshot",
- _unexpected_upload,
- )
- monkeypatch.setattr(
- "engraphis.cloud_features.CloudFeatureClient.from_environment",
- lambda workspace_id=None: _Cloud(),
- )
- with _client(monkeypatch, tmp_path) as client:
- assert client.get("/api/automation").status_code == 200
- response = client.post("/api/automation", json={"enabled": False})
- assert response.status_code == 200
- assert saved["enabled"] is False
-
-
-def test_automation_and_maintenance_use_the_selected_workspace(monkeypatch, tmp_path):
- policy_workspaces = []
- snapshot_workspaces = []
- maintenance_workspaces = []
-
- class _Cloud:
- def get_policy(self, workspace_id):
- policy_workspaces.append(workspace_id)
- return {"enabled": False, "cadence_minutes": 60, "dream_enabled": True}
-
- def list_jobs(self, workspace_id, *, limit=10):
- policy_workspaces.append(workspace_id)
- return {"jobs": []}
-
- def upload_snapshot(self, workspace_id, snapshot):
- snapshot_workspaces.append(workspace_id)
- return {"generation": snapshot["generation"]}
-
- def save_policy(self, workspace_id, policy):
- policy_workspaces.append(workspace_id)
- return {"version": 1}
-
- def snapshot(service, workspace):
- snapshot_workspaces.append(workspace)
- return service._lookup_workspace(workspace), {"generation": 1}
-
- def managed_job(service, workspace, kind):
- maintenance_workspaces.append((workspace, kind))
- return {"result": {"kind": kind}}
-
- monkeypatch.setattr("engraphis.cloud_features.build_managed_snapshot", snapshot)
- monkeypatch.setattr("engraphis.cloud_features.run_managed_job", managed_job)
- monkeypatch.setattr(
- "engraphis.cloud_features.CloudFeatureClient.from_environment",
- lambda workspace_id=None: _Cloud(),
- )
- with _client(monkeypatch, tmp_path) as client:
- beta_id = client.app.state.service._lookup_workspace("beta")
- demo_id = client.app.state.service._lookup_workspace("demo")
- assert client.get("/api/automation?workspace=beta").status_code == 200
- assert client.post(
- "/api/automation?workspace=beta", json={"enabled": True}
- ).status_code == 200
- assert client.post(
- "/api/maintenance/run?workspace=beta", json={"dry_run": True}
- ).status_code == 200
-
- assert beta_id in policy_workspaces
- assert demo_id not in policy_workspaces
- assert "beta" in snapshot_workspaces
- assert maintenance_workspaces == [("beta", "consolidate")]
-
-
-def test_automation_workspace_query_unknown_is_not_replaced_by_legacy_default(
- monkeypatch, tmp_path
-):
- with _client(monkeypatch, tmp_path) as client:
- for method, path, payload in (
- (client.get, "/api/automation?workspace=missing", None),
- (client.post, "/api/automation?workspace=missing", {"enabled": False}),
- (client.post, "/api/maintenance/run?workspace=missing", {"dry_run": True}),
- ):
- response = method(path, json=payload) if payload is not None else method(path)
- assert response.status_code == 404
-
-
-def test_dashboard_automation_uses_active_workspace_and_discloses_upload_boundary():
- source = Path(__file__).parents[1] / "engraphis" / "static" / "dashboard.js"
- source = source.read_text(encoding="utf-8")
- assert "/automation?workspace=" in source
- assert "/maintenance/run?workspace=" in source
- assert "Preview snapshot" not in source
- assert "uploads the selected workspace’s normal and sensitive memory content" in source
- # The upload boundary is still disclosed, but consent now travels with the cloud
- # account: the dashboard must not name the operator override anywhere.
- assert "ENGRAPHIS_MANAGED_COMPUTE_CONSENT" not in source
- assert "Hosted work is automatic with Pro." in source
-
-
-def test_portfolio_and_report_analytics_are_hosted_only(monkeypatch, tmp_path):
- with _client(monkeypatch, tmp_path) as client:
- assert client.get("/api/analytics/portfolio").status_code == 501
- assert client.get("/api/analytics/export?workspace=demo").status_code == 501
-
-
-def test_raw_owner_export_is_free_and_signed_export_is_honestly_unimplemented(
- monkeypatch, tmp_path
-):
- """The signed variant must not claim to exist somewhere else.
-
- It previously answered ``cloud_only: True`` — but Engraphis Cloud has no export route,
- no supported hosted export capability, so that pointed a customer at a
- product that does not exist. The 501 now says the capability is unimplemented and names
- the working unsigned export instead.
- """
-
- with _client(monkeypatch, tmp_path) as client:
- raw = client.get("/api/export?workspace=demo")
- assert raw.status_code == 200
- assert raw.json()["counts"]["memories"] >= 1
- signed = client.get("/api/export?workspace=demo&signed=true")
- assert signed.status_code == 501
- detail = signed.json()["detail"]
- assert detail["implemented"] is False
- assert detail["alternative"] == "/export"
- assert "cloud_only" not in detail
- assert "Engraphis Cloud" not in detail["error"]
-
-
-def test_health_and_readiness_remain_public(monkeypatch, tmp_path):
- with _client(monkeypatch, tmp_path) as client:
- assert client.get("/api/health").status_code == 200
- assert client.get("/api/ready").status_code == 200
-
-
-def test_dashboard_exception_responses_do_not_echo_untrusted_exception_text():
- secret = "https://provider.example/?api_key=do-not-return-this"
-
- def fail_with(exc):
- raise exc
-
- with pytest.raises(HTTPException) as internal:
- v2_api._run(fail_with, RuntimeError(secret))
- assert internal.value.status_code == 500
- assert internal.value.detail == {"error": "internal server error"}
- assert secret not in repr(internal.value.detail)
-
- with pytest.raises(HTTPException) as validation:
- v2_api._run(fail_with, ValidationError(secret))
- assert validation.value.status_code == 400
- assert validation.value.detail == {"error": "invalid request"}
- assert secret not in repr(validation.value.detail)
-
- with pytest.raises(HTTPException) as downstream:
- v2_api._run(fail_with, HTTPException(status_code=418, detail={"error": secret}))
- assert downstream.value.status_code == 418
- assert downstream.value.detail == {"error": "request rejected"}
- assert secret not in repr(downstream.value.detail)
-
- with pytest.raises(HTTPException) as invalid_status:
- v2_api._run(fail_with, HTTPException(status_code=999, detail={"error": secret}))
- assert invalid_status.value.status_code == 500
- assert invalid_status.value.detail == {"error": "internal server error"}
- assert secret not in repr(invalid_status.value.detail)
-
- with pytest.raises(HTTPException) as mismatch:
- v2_api._run(fail_with, ValueError(f"{secret}: shapes 256 and 384 are not aligned"))
- assert mismatch.value.status_code == 409
- assert mismatch.value.detail["embedder"] is True
- assert secret not in repr(mismatch.value.detail)
-
- with pytest.raises(HTTPException) as ordinary_value_error:
- v2_api._run(fail_with, ValueError(secret))
- assert ordinary_value_error.value.status_code == 400
- assert ordinary_value_error.value.detail == {"error": "invalid request"}
- assert secret not in repr(ordinary_value_error.value.detail)
-
-
-def test_dashboard_engine_value_error_is_a_sanitized_client_error(monkeypatch, tmp_path):
- secret = "malformed document details must stay private"
- with _client(monkeypatch, tmp_path) as client:
- def reject_document(*_args, **_kwargs):
- raise ValueError(secret)
-
- monkeypatch.setattr(client.app.state.service, "remember", reject_document)
- response = client.post(
- "/api/remember",
- json={"content": "client document", "workspace": "demo"},
- )
-
- assert response.status_code == 400
- assert response.json() == {"detail": {"error": "invalid request"}}
- assert secret not in response.text
-
-
-def test_managed_cloud_errors_forward_only_bounded_public_copy():
- """``_managed_call`` forwards the message; the bound is the boundary's own check.
-
- ``CloudFeatureError`` is the already-redacted form -- every raise site builds it from
- fixed, status-keyed copy -- so its text is what the customer should read. The bound
- here is not the redaction, it is the guard for a message that is *not* that fixed copy:
- anything oversized, empty, or carrying control characters is dropped for the generic
- placeholder rather than rendered into a JSON error body.
- """
-
- def fail_with(exc):
- raise exc
-
- for message in ("x" * 301, "", "connection\x00reset", "trace\x1b[31m"):
- with pytest.raises(HTTPException) as caught:
- v2_api._managed_call(fail_with, CloudFeatureError(message, status=502))
- assert caught.value.status_code == 502
- assert caught.value.detail == {
- "error": v2_api._MANAGED_ERROR_FALLBACK, "managed_cloud": True,
- "transient": False,
- }
-
- with pytest.raises(HTTPException) as consent:
- v2_api._managed_call(
- fail_with,
- CloudFeatureError(
- "Managed compute is turned off for this installation.",
- status=409, code="consent_required",
- ),
- )
- assert consent.value.status_code == 409
- assert consent.value.detail == {
- "error": "Managed compute is turned off for this installation.",
- "managed_cloud": True,
- "transient": False,
- "code": "consent_required",
- }
-
- with pytest.raises(HTTPException) as unconfigured:
- v2_api._managed_call(
- fail_with,
- CloudFeatureError(
- "Connect this installation to Engraphis Cloud to use hosted features.",
- status=401, code="cloud_unconfigured",
- ),
- )
- assert unconfigured.value.status_code == 401
- assert unconfigured.value.detail == {
- "error": "Connect this installation to Engraphis Cloud to use hosted features.",
- "managed_cloud": True,
- "transient": False,
- "code": "cloud_unconfigured",
- }
-
-
-@pytest.mark.parametrize("status", (401, 402, 403))
-def test_managed_authorization_denial_settles_local_entitlement(monkeypatch, status):
- """A live hosted denial must immediately retire stale paid presentation state."""
-
- calls = []
- monkeypatch.setattr(v2_api, "_record_authoritative_denial", lambda: calls.append(status))
-
- def fail_with(exc):
- raise exc
-
- with pytest.raises(HTTPException) as caught:
- v2_api._managed_call(
- fail_with, CloudFeatureError("Engraphis Cloud authorization was rejected.",
- status=status),
- )
-
- assert caught.value.status_code == status
- assert calls == [status]
-
-
-@pytest.mark.parametrize("status", (409, 429, 503))
-def test_managed_non_authorization_failures_do_not_settle_entitlement(monkeypatch, status):
- """Conflicts and outages do not prove that a subscription or membership changed."""
-
- calls = []
- monkeypatch.setattr(v2_api, "_record_authoritative_denial", lambda: calls.append(status))
-
- def fail_with(exc):
- raise exc
-
- with pytest.raises(HTTPException):
- v2_api._managed_call(
- fail_with, CloudFeatureError("Engraphis Cloud temporarily failed.", status=status),
- )
-
- assert calls == []
-
-
-def _managed_http_failure(monkeypatch, status: int) -> HTTPException:
- """Drive one real hosted request against a control plane that answers ``status``."""
-
- class _Opener:
- def open(self, request, timeout=None):
- raise urllib.error.HTTPError(
- "https://compute.example.test/private", status, "failure", {},
- io.BytesIO(b'{"detail": "provider-internals https://backend.invalid"}'),
- )
-
- monkeypatch.setattr(
- cloud_features, "build_pinned_https_opener", lambda *handlers: _Opener()
- )
- client = cloud_features.CloudFeatureClient(
- "https://compute.example.test", "org_1", "token"
- )
- with pytest.raises(HTTPException) as caught:
- v2_api._managed_call(client._request, "GET", "/private")
- return caught.value
-
-
-def test_a_managed_outage_is_distinguishable_from_a_workspace_conflict(monkeypatch):
- """The defect: every hosted failure rendered as one fixed, unactionable string.
-
- ``cloud_features._public_http_error`` already produces redacted, status-keyed copy that
- tells a retryable outage apart from a conflict the customer has to fix -- and
- ``_managed_call`` threw all of it away, so the dashboard's error branch could only ever
- show "managed cloud operation failed" for a 429, a 5xx and a 409 alike.
- """
-
- busy = _managed_http_failure(monkeypatch, 429)
- down = _managed_http_failure(monkeypatch, 503)
- conflict = _managed_http_failure(monkeypatch, 409)
-
- assert busy.status_code == 429
- assert busy.detail["transient"] is True
- assert "temporarily busy" in busy.detail["error"], busy.detail["error"]
-
- assert down.status_code == 503
- assert down.detail["transient"] is True
- assert "temporarily unavailable" in down.detail["error"], down.detail["error"]
-
- assert conflict.status_code == 409
- assert conflict.detail["transient"] is False
- assert "workspace state" in conflict.detail["error"], conflict.detail["error"]
-
- messages = {busy.detail["error"], down.detail["error"], conflict.detail["error"]}
- assert len(messages) == 3, "the dashboard still cannot tell these three apart"
- assert v2_api._MANAGED_ERROR_FALLBACK not in messages
- # Forwarding the public copy must not forward the provider's body with it.
- assert all("provider-internals" not in text for text in messages)
- assert all("backend.invalid" not in text for text in messages)
-
-
-def test_every_managed_cloud_error_message_is_fixed_local_copy():
- """The invariant that makes forwarding safe, pinned against future raise sites.
-
- ``_managed_call`` may forward a ``CloudFeatureError`` message only because every one of
- them is built from a literal in this repository -- never from a provider body, a
- ``CloudSessionError``, or a local path. A raise site that interpolated a runtime value
- would silently turn this boundary into a reflection point, so the shape is asserted
- rather than trusted.
-
- Three forms are accepted: a string literal; a name bound from ``_public_http_error`` /
- ``_public_session_error`` (both of which switch on a bare integer status and return
- fixed copy); and the one audited ``%`` template, below.
- """
-
- source = Path(cloud_features.__file__).read_text(encoding="utf-8")
- tree = ast.parse(source)
-
- public_copy = {"_public_http_error", "_public_session_error"}
- from_public_copy = set()
- for node in ast.walk(tree):
- if not isinstance(node, ast.Assign) or not isinstance(node.value, ast.Call):
- continue
- called = node.value.func
- if not isinstance(called, ast.Name) or called.id not in public_copy:
- continue
- for target in node.targets:
- elements = target.elts if isinstance(target, ast.Tuple) else [target]
- from_public_copy.update(
- item.id for item in elements if isinstance(item, ast.Name)
- )
- assert from_public_copy, "the fixed-copy helpers are no longer bound to a name"
-
- interpolated = []
- for node in ast.walk(tree):
- if not isinstance(node, ast.Call):
- continue
- name = node.func.id if isinstance(node.func, ast.Name) else None
- if name != "CloudFeatureError" or not node.args:
- continue
- message = node.args[0]
- if isinstance(message, ast.Constant) and isinstance(message.value, str):
- continue
- if isinstance(message, ast.Name) and message.id in from_public_copy:
- continue
- # ``"literal %s" % (...)`` is allowed only where the substituted values are
- # themselves constrained to local literals; ``run_job`` is the single such site
- # and its ``status`` is guarded by an ``in {"failed", "canceled"}`` membership
- # test one line above. Anything else -- an f-string, a bare name, a concatenated
- # response field -- is a reflection risk and fails here.
- if (isinstance(message, ast.BinOp) and isinstance(message.op, ast.Mod)
- and isinstance(message.left, ast.Constant)
- and message.left.value == "Managed %s did not complete (%s)."):
- continue
- interpolated.append((node.lineno, ast.dump(message)[:120]))
-
- assert interpolated == [], (
- "a CloudFeatureError message is no longer fixed local copy; _managed_call "
- "forwards it to the customer: %r" % (interpolated,)
- )
+"""Unified local dashboard tests for the public open-core boundary."""
+import ast
+import io
+import threading
+import urllib.error
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
+
+import pytest
+
+pytest.importorskip("fastapi", reason="full-stack extra not installed")
+pytest.importorskip("httpx", reason="httpx not installed")
+
+from fastapi.testclient import TestClient # noqa: E402
+from fastapi import HTTPException # noqa: E402
+
+from engraphis import cloud_features # noqa: E402
+from engraphis.config import settings # noqa: E402
+from engraphis.cloud_features import CloudFeatureError # noqa: E402
+from engraphis.core.interfaces import MemoryType, Scope # noqa: E402
+from engraphis.routes import v2_api # noqa: E402
+from engraphis.service import MemoryService, ValidationError # noqa: E402
+
+
+def _client(monkeypatch, tmp_path):
+ db_path = str(tmp_path / "dashboard.db")
+ monkeypatch.setattr(settings, "db_path", db_path)
+ monkeypatch.setattr(settings, "embed_model", "")
+ monkeypatch.setattr(settings, "embed_dim", 384)
+ monkeypatch.setattr(settings, "allowed_workspaces", [])
+ monkeypatch.setattr(settings, "api_token", "")
+ seeded = MemoryService.create(db_path)
+ demo_id = seeded.store.get_or_create_workspace("demo")
+ beta_id = seeded.store.get_or_create_workspace("beta")
+ seeded.engine.remember(
+ "Postgres 16 is the main database.",
+ workspace_id=demo_id,
+ scope=Scope.WORKSPACE,
+ title="Database",
+ )
+ seeded.engine.remember(
+ "A second workspace must stay isolated.",
+ workspace_id=beta_id,
+ scope=Scope.WORKSPACE,
+ title="Isolation",
+ )
+ seeded.store.close()
+ from engraphis.dashboard_app import create_app
+ return TestClient(create_app(), client=("127.0.0.1", 50000))
+
+
+def test_dashboard_serves_and_bootstraps_local_core(monkeypatch, tmp_path):
+ with _client(monkeypatch, tmp_path) as client:
+ page = client.get("/")
+ assert page.status_code == 200
+ assert "
Engraphis Ledger" in page.text
+ assert 'class="sidebar"' in page.text
+ for area in ("Today", "Ask", "Library", "Graph & Relationships", "Provenance", "Manage"):
+ assert f">{area}<" in page.text
+ assert 'value="matrix">Matrix' in page.text
+ assert 'class="dashboard-switcher" aria-label="Dashboard interface"' in page.text
+ assert 'id="sidebar-theme-select" aria-label="Dashboard theme"' in page.text
+ assert 'value="classic">Classic<' in page.text
+ assert 'href="/classic">Classic<' in page.text
+ assert 'Ledger (primary)' not in page.text
+ assert 'Classic (alternate)' not in page.text
+ assert '/v2-assets/vendor/d3.min.js' in page.text
+ assert '/v2-assets/vendor/force-graph.min.js' not in page.text
+ assert '/v2-assets/engraphis-graph.js' not in page.text
+ classic = client.get("/classic")
+ assert classic.status_code == 200
+ assert '/classic-assets/dashboard.css' in classic.text
+ assert 'class="dashboard-switcher" aria-label="Dashboard interface"' in classic.text
+ assert 'href="/"' in classic.text
+ assert 'href="/classic" aria-current="page">Classic (alternate)<' in classic.text
+ assert 'value="classic" selected>Classic dashboard (alternate)<' in classic.text
+ assert 'id="graph-show-all"' not in classic.text
+ assert client.get("/v2-assets/ledger.css").status_code == 200
+ ledger_js = client.get("/v2-assets/ledger.js")
+ assert ledger_js.status_code == 200
+ assert "'/v2-assets/vendor/force-graph.min.js?v=20260727-final'" in ledger_js.text
+ assert "'/v2-assets/engraphis-graph.js?v=20260730-drag-stability'" in ledger_js.text
+ assert "/v2-assets/ledger.css?v=20260728-connected-memories" in page.text
+ assert "/v2-assets/ledger.js?v=20260728-connected-memories" in page.text
+ classic_js = client.get("/classic-assets/dashboard.js")
+ assert classic_js.status_code == 200
+ assert "/static/vendor/force-graph.min.js" in classic_js.text
+ assert "/v2-assets/engraphis-graph.js?v=20260730-drag-stability" in classic_js.text
+ assert "graphLimit=GRAPH_FULL?20000:320" in classic_js.text
+ assert "graphScope=GRAPH_FULL?'&full=true':(showUnlinked?'':'&connected_only=true')" in classic_js.text
+ bootstrap = client.get("/api/bootstrap")
+ assert bootstrap.status_code == 200
+ assert bootstrap.json()["stats"]["memories"] >= 1
+ savings = client.get("/api/context-savings", params={"workspace": "demo"})
+ assert savings.status_code == 200
+ assert savings.json()["format"] == "engraphis-context-savings/1"
+ filtered = client.get(
+ "/api/context-savings",
+ params={"workspace": "demo", "from_ts": 0, "to_ts": 9_999_999_999,
+ "release_version": "1.5"},
+ )
+ assert filtered.status_code == 200
+ assert filtered.json()["period"] == {"from_ts": 0, "to_ts": 9_999_999_999}
+ assert "Estimated context saved" in page.text
+
+
+def test_dashboard_assets_revalidate_instead_of_pinning_old_visuals(monkeypatch, tmp_path):
+ with _client(monkeypatch, tmp_path) as client:
+ for path in (
+ "/v2-assets/engraphis-graph.js?v=20260730-drag-stability",
+ "/v2-assets/ledger.js?v=20260728-connected-memories",
+ "/v2-assets/ledger.css?v=20260728-connected-memories",
+ "/classic-assets/dashboard.js?v=20260728-reference-materials",
+ ):
+ response = client.get(path)
+ assert response.status_code == 200
+ assert response.headers["cache-control"] == "no-cache, must-revalidate"
+
+
+def test_classic_dashboard_script_mirrors_the_static_compatibility_asset():
+ root = Path(__file__).parents[1] / "engraphis"
+ assert (root / "classic_assets" / "dashboard.js").read_bytes() == (
+ root / "static" / "dashboard.js"
+ ).read_bytes()
+
+
+def test_dashboard_and_mcp_recall_share_the_v2_service(monkeypatch, tmp_path):
+ pytest.importorskip("mcp", reason="MCP extra not installed")
+ import json
+
+ from engraphis import mcp_server
+
+ with _client(monkeypatch, tmp_path) as client:
+ assert mcp_server.service() is client.app.state.service
+ response = client.get(
+ "/api/recall",
+ params={"q": "which database do we use", "workspace": "demo", "k": 3},
+ )
+ assert response.status_code == 200
+ dashboard = response.json()
+ mcp = json.loads(mcp_server.engraphis_recall(
+ query="which database do we use", workspace="demo", k=3,
+ ))
+ assert [memory["id"] for memory in dashboard["memories"]] == [
+ memory["id"] for memory in mcp["memories"]
+ ]
+ assert [memory["retention"] for memory in dashboard["memories"]] == [
+ memory["retention"] for memory in mcp["memories"]
+ ]
+ assert [memory["relative_score"] for memory in dashboard["memories"]] == [
+ memory["relative_score"] for memory in mcp["memories"]
+ ]
+ assert [memory["absolute_support"] for memory in dashboard["memories"]] == [
+ memory["absolute_support"] for memory in mcp["memories"]
+ ]
+ assert dashboard["score_semantics"] == mcp["score_semantics"]
+
+
+def test_dashboard_keyword_fallback_reports_truthful_lexical_scores(monkeypatch, tmp_path):
+ with _client(monkeypatch, tmp_path) as client:
+ def mismatched_embedder(*_args, **_kwargs):
+ raise ValueError("shapes (1,256) and (384,1) not aligned")
+
+ monkeypatch.setattr(client.app.state.service, "recall", mismatched_embedder)
+ response = client.get(
+ "/api/recall",
+ params={
+ "q": "which database do we use",
+ "workspace": "demo",
+ "k": 3,
+ "response_mode": "compact",
+ },
+ )
+
+ assert response.status_code == 200
+ payload = response.json()
+ assert payload["mode"] == "keyword"
+ assert "lexical Jaccard" in payload["score_semantics"]["relative_score"]
+ assert "Semantic support is unavailable" in (
+ payload["score_semantics"]["absolute_support"]
+ )
+ memory = payload["memories"][0]
+ assert memory["score"] == memory["relative_score"] == 1.0
+ assert 0.0 < memory["absolute_support"] < 1.0
+ assert memory["arm"] == "lexical"
+ assert "content" not in memory
+
+
+def test_dashboard_keyword_fallback_applies_requested_memory_type_limits(
+ monkeypatch, tmp_path
+):
+ with _client(monkeypatch, tmp_path) as client:
+ workspace_id = client.app.state.service.store.get_or_create_workspace("demo")
+ client.app.state.service.engine.remember(
+ "Database upgrade procedure requires a verified backup.",
+ workspace_id=workspace_id,
+ scope=Scope.WORKSPACE,
+ mtype=MemoryType.PROCEDURAL,
+ title="Database procedure",
+ )
+
+ def mismatched_embedder(*_args, **_kwargs):
+ raise ValueError("shapes (1,256) and (384,1) not aligned")
+
+ monkeypatch.setattr(client.app.state.service, "recall", mismatched_embedder)
+ response = client.get(
+ "/api/recall",
+ params={
+ "q": "database",
+ "workspace": "demo",
+ "k": 3,
+ "mtype_limits": '{"semantic":0,"procedural":1}',
+ },
+ )
+
+ assert response.status_code == 200
+ payload = response.json()
+ assert payload["mtype_limits"] == {"semantic": 0, "procedural": 1}
+ assert [memory["memory_type"] for memory in payload["memories"]] == [
+ "procedural"
+ ]
+
+
+@pytest.mark.parametrize("invalid_limit", [True, "2"])
+def test_dashboard_post_recall_surfaces_reject_coerced_memory_type_limits(
+ monkeypatch, tmp_path, invalid_limit
+):
+ with _client(monkeypatch, tmp_path) as client:
+ response = client.post(
+ "/api/intent/recall",
+ json={"query": "database", "mtype_limits": {"semantic": invalid_limit}},
+ )
+
+ assert response.status_code == 422
+
+
+def test_dashboard_serves_the_graph_engine_from_its_v2_asset_surface(monkeypatch, tmp_path):
+ with _client(monkeypatch, tmp_path) as client:
+ asset = client.get("/v2-assets/engraphis-graph.js")
+ assert asset.status_code == 200
+ assert "window.EngraphisGraph =" in asset.text
+ compat = client.get("/v2-assets/engraphis-graph-compat.js")
+ assert compat.status_code == 200
+ assert "window.EngraphisGraphCompat =" in compat.text
+ assert client.get("/v2-assets/vendor/d3.min.js").status_code == 200
+ assert client.get("/v2-assets/vendor/force-graph.min.js").status_code == 200
+
+
+def test_graph_load_is_bounded_single_flight_and_retryable(monkeypatch, tmp_path):
+ with _client(monkeypatch, tmp_path) as client:
+ page = client.get("/")
+ script = client.get("/v2-assets/ledger.js")
+ assert 'id="graph-retry"' in page.text
+ assert 'id="graph-full"' not in page.text
+ assert '>Show all nodes<' not in page.text
+ assert 'id="graph-show-unlinked"' in page.text
+ assert 'id="graph-unlinked"' not in page.text
+ assert 'id="graph-tune-unlinked"' not in page.text
+ assert 'id="graph-style" type="hidden" value="cyber"' in page.text
+ assert "const GRAPH_INITIAL_NODE_LIMIT = 320;" in script.text
+ assert "const GRAPH_FULL_NODE_LIMIT = 20_000;" in script.text
+ assert "const GRAPH_LOAD_TIMEOUT_MS = 12_000;" in script.text
+ assert "AbortController" in script.text
+ assert "state.graphLoadPromise" in script.text
+ assert "&full=true" in script.text
+ assert "&connected_only=true" in script.text
+ assert "style: 'cyber'" in script.text
+ assert "renderMode: targetMode" in script.text
+ assert "loadGraph({ force: true })" in script.text
+
+
+def test_graph_motion_saved_views_and_tuning_controls_are_wired(monkeypatch, tmp_path):
+ with _client(monkeypatch, tmp_path) as client:
+ page = client.get("/")
+ script = client.get("/v2-assets/ledger.js")
+ for control in (
+ 'id="graph-flow-speed"', 'data-graph-saved-view="operations"',
+ 'data-graph-saved-view="schema"', 'data-graph-saved-view="people"',
+ 'data-graph-saved-view="code"', 'id="graph-save-view"',
+ 'id="graph-repel"', 'id="graph-depth"', 'id="graph-reset-tuning"',
+ 'data-graph-layer="code"',
+ ):
+ assert control in page.text
+ for behavior in (
+ "function applyGraphView(id)", "function resetGraphTuning()",
+ "function saveCurrentGraphView()", "function graphTuningSettings()",
+ "&include_code=true", "graph.setLayers(graphLayerState())",
+ "setSettings({ flowSpeed: speed })",
+ ):
+ assert behavior in script.text
+
+
+def test_graph_palette_recolors_every_colour_mode(monkeypatch, tmp_path):
+ with _client(monkeypatch, tmp_path) as client:
+ engine = client.get("/v2-assets/engraphis-graph.js")
+ ledger = client.get("/v2-assets/ledger.js")
+ assert engine.status_code == 200
+ assert "function selectedPalette()" in engine.text
+ assert "function commPal() {" in engine.text
+ assert "return selectedPalette() ||" in engine.text
+ assert "const colors = selectedPalette() || GRAPH_HEAT;" in engine.text
+ # Palettes still recolor every identity mode, but material families stay stable:
+ # semantic color belongs to the slim identity ring rather than rotating the whole
+ # Cyber film into arbitrary green/yellow alloys.
+ assert "function iridescentTint(c)" not in engine.text
+ assert "fixedPalette" in engine.text
+ assert "function identityRing(" in engine.text
+ assert "identity: rgbString(identity)" in engine.text
+ assert "function graphThemeColors()" in ledger.text
+ assert "graph.setThemeColors(graphThemeColors());" in ledger.text
+ assert "state.graphEngine.setThemeColors(graphThemeColors());" in ledger.text
+ assert "renderMode: opts.renderMode === 'full' ? 'full' : 'overview'" in engine.text
+ assert "function pinFullGraphLayout(data)" in engine.text
+
+
+def test_graph_facts_and_search_use_the_atomic_node_reveal(monkeypatch, tmp_path):
+ with _client(monkeypatch, tmp_path) as client:
+ page = client.get("/")
+ ledger = client.get("/v2-assets/ledger.js")
+ engine = client.get("/v2-assets/engraphis-graph.js")
+ assert 'id="graph-connections-dialog"' in page.text
+ assert "function revealGraphNode(id, label = 'Selected entity')" in ledger.text
+ assert "revealGraphNode(item.id, item.name)" in ledger.text
+ assert "function openGraphConnections(item)" in ledger.text
+ assert "function showGraphConnectionMemories(item)" in ledger.text
+ assert "onNodeClick: item => openGraphConnections(item)" in ledger.text
+ assert "api.reveal = id =>" in engine.text
+ assert "function centerRenderedNode(id)" in engine.text
+ assert "suppressNodeClickAfterDrag" in engine.text
+ assert "render(true, true);" not in engine.text[engine.text.index("api.focus = id =>"):engine.text.index("api.clearFocus")]
+
+
+def test_library_editor_stacks_directly_below_the_selected_memory_panel(monkeypatch, tmp_path):
+ with _client(monkeypatch, tmp_path) as client:
+ page = client.get("/")
+ assert page.status_code == 200
+ assert '
' in page.text
+ assert page.text.index('id="memory-detail"') < page.text.index('id="memory-editor"')
+ stylesheet = client.get("/v2-assets/ledger.css")
+ assert ".library-detail-stack { display: grid; gap: 12px; align-content: start; }" in stylesheet.text
+
+
+def test_workspace_switcher_uses_the_active_ledger_theme_for_native_dropdowns(monkeypatch, tmp_path):
+ with _client(monkeypatch, tmp_path) as client:
+ stylesheet = client.get("/v2-assets/ledger.css")
+ assert stylesheet.status_code == 200
+ css = stylesheet.text
+ assert ".workspace-switcher select {" in css
+ assert "background: var(--c-inset);" in css
+ assert "color-scheme: dark;" in css
+ assert 'body[data-theme="paper"] .workspace-switcher select { color-scheme: light; }' in css
+ assert ".workspace-switcher select option { background: var(--c-inset); color: var(--c-fg); }" in css
+ assert ".workspace-switcher select option:checked { background: var(--c-acc); color: var(--c-bg); }" in css
+
+
+def test_sidebar_keeps_manage_and_compare_plans_in_separate_flex_rows(monkeypatch, tmp_path):
+ with _client(monkeypatch, tmp_path) as client:
+ stylesheet = client.get("/v2-assets/ledger.css")
+ assert stylesheet.status_code == 200
+ css = stylesheet.text
+ sidebar = css[css.index(".sidebar {"):css.index(".brand-row {")]
+ assert "display: flex;" in sidebar
+ assert "flex-direction: column;" in sidebar
+ assert "grid-template-rows" not in sidebar
+ assert ".primary-nav { flex: 1 0 auto; }" in css
+ assert ".manage-nav { flex: 0 0 auto; }" in css
+ assert ".sidebar-promo {\n flex: 0 0 auto;" in css
+
+
+def test_dashboard_grounded_answer_route_cites_or_abstains(monkeypatch, tmp_path):
+ with _client(monkeypatch, tmp_path) as client:
+ grounded = client.post(
+ "/api/answer",
+ json={
+ "query": "Which database is the main database?",
+ "workspace": "demo",
+ "k": 8,
+ "max_citations": 5,
+ "candidate_depth": "adaptive",
+ },
+ )
+ assert grounded.status_code == 200
+ body = grounded.json()
+ assert body["query"] == "Which database is the main database?"
+ assert body["grounded"] is True
+ assert body["abstained"] is False
+ assert body["citations"]
+ assert body["sources"] == body["citations"]
+ assert "[1]" in body["answer"]
+ assert body["candidate_depth"] == "adaptive"
+ # ``candidate_k_used`` is the final page depth after prompt-safe
+ # overfetch/widening, rather than the adaptive policy's starting depth.
+ assert body["candidate_k_used"] >= body["candidate_k_requested"]
+
+ abstained = client.post(
+ "/api/answer",
+ json={
+ "query": "How should I bake a sourdough loaf?",
+ "workspace": "demo",
+ },
+ )
+ assert abstained.status_code == 200
+ assert abstained.json()["grounded"] is False
+ assert abstained.json()["abstained"] is True
+
+
+def test_dashboard_grounded_answer_route_bounds_and_redacts(monkeypatch, tmp_path):
+ with _client(monkeypatch, tmp_path) as client:
+ assert client.post("/api/answer", json={"query": "", "workspace": "demo"}).status_code == 422
+ assert client.post(
+ "/api/answer",
+ json={"query": "database", "workspace": "demo", "k": 51},
+ ).status_code == 422
+
+
+def test_team_account_routes_are_not_in_public_runtime(monkeypatch, tmp_path):
+ with _client(monkeypatch, tmp_path) as client:
+ assert client.post("/api/auth/setup", json={}).status_code == 404
+ assert client.get("/api/auth/users").status_code == 404
+ state = client.get("/api/auth/state").json()
+ assert state["enabled"] is False
+ assert state["hosted_team"] is True
+
+
+def test_local_agent_write_has_no_client_side_team_paywall(monkeypatch, tmp_path):
+ with _client(monkeypatch, tmp_path) as client:
+ response = client.post(
+ "/api/remember",
+ json={"workspace": "demo", "content": "Queues use at-least-once delivery."},
+ )
+ assert response.status_code == 200
+
+
+def test_http_memory_api_exposes_world_timed_agent_writes_immediately(monkeypatch, tmp_path):
+ with _client(monkeypatch, tmp_path) as client:
+ old = client.post(
+ "/api/remember",
+ json={
+ "workspace": "demo",
+ "content": "The API rate limit is 100 requests per minute.",
+ "valid_from": 1_000.0,
+ "subject_key": "api.rate_limit",
+ "claim_kind": "configured_value",
+ },
+ ).json()
+ new = client.post(
+ "/api/intent/remember",
+ json={
+ "workspace": "demo",
+ "text": "The API rate limit is 500 requests per minute.",
+ "valid_from": 2_000.0,
+ "subject_key": "api.rate_limit",
+ "claim_kind": "configured_value",
+ },
+ ).json()
+
+ before = client.get(
+ "/api/recall",
+ params={
+ "workspace": "demo",
+ "q": "What is the API rate limit?",
+ "as_of": 1_500.0,
+ },
+ )
+ after = client.post(
+ "/api/answer",
+ json={
+ "workspace": "demo",
+ "query": "What is the API rate limit?",
+ "as_of": 2_500.0,
+ "min_support": 0.0,
+ },
+ )
+
+ assert before.status_code == 200
+ assert [memory["id"] for memory in before.json()["memories"]] == [old["id"]]
+ assert after.status_code == 200
+ assert after.json()["sources"]
+ service = client.app.state.service
+ assert service.store.get_memory(old["id"]).valid_from == 1_000.0
+ assert service.store.get_memory(new["id"]).valid_from == 2_000.0
+ assert service.store.get_memory(old["id"]).provenance["review_state"] == "approved"
+ assert service.store.get_memory(new["id"]).provenance["review_state"] == "approved"
+
+
+def test_keyword_recall_fallback_keeps_bitemporal_visibility(monkeypatch, tmp_path):
+ """A semantic-backend failure must not leak current facts into historical views."""
+ with _client(monkeypatch, tmp_path) as client:
+ svc = v2_api.service()
+ workspace_id = svc.store.get_or_create_workspace("demo")
+ old = {"id": svc.engine.remember(
+ "The fallback retention setting was ten days.", workspace_id=workspace_id,
+ scope=Scope.WORKSPACE, valid_from=1_000.0, resolve_conflicts=False,
+ )}
+ new = {"id": svc.engine.remember(
+ "The fallback retention setting was thirty days.", workspace_id=workspace_id,
+ scope=Scope.WORKSPACE, valid_from=2_000.0, resolve_conflicts=False,
+ )}
+ # The writes happened during this test, but the fixture models facts learned
+ # before the requested historical system-time anchors.
+ svc.store.conn.execute(
+ "UPDATE memories SET ingested_at=100 WHERE id=?", (old["id"],)
+ )
+ svc.store.conn.execute(
+ "UPDATE memories SET ingested_at=200 WHERE id=?", (new["id"],)
+ )
+ svc.store.conn.execute(
+ "UPDATE memories SET valid_to=2000, valid_to_recorded_at=200, "
+ "subject_key='retention.days', claim_kind='configured_value' "
+ "WHERE id=?",
+ (old["id"],),
+ )
+ svc.store.conn.commit()
+ old_before = v2_api._keyword_search(
+ "demo", "fallback retention", valid_at=1_500.0, known_at=3_000.0
+ )
+ old_known = v2_api._keyword_search(
+ "demo", "fallback retention", valid_at=1_500.0, known_at=50.0
+ )
+ current = v2_api._keyword_search(
+ "demo", "fallback retention", valid_at=2_500.0, known_at=3_000.0
+ )
+ closure_unknown = v2_api._keyword_search(
+ "demo", "fallback retention", valid_at=2_500.0, known_at=150.0
+ )
+
+ assert [memory["id"] for memory in old_before] == [old["id"]]
+ assert old_known == []
+ assert [memory["id"] for memory in current] == [new["id"]]
+ assert [memory["id"] for memory in closure_unknown] == [old["id"]]
+ assert closure_unknown[0]["valid_to_recorded_at"] == 200.0
+ assert closure_unknown[0]["subject_key"] == "retention.days"
+ assert closure_unknown[0]["claim_kind"] == "configured_value"
+
+ def incompatible_embedder(*_args, **_kwargs):
+ raise ValueError("shapes (256,) and (384,) not aligned")
+
+ monkeypatch.setattr(svc, "recall", incompatible_embedder)
+ fallback = client.get(
+ "/api/recall",
+ params={
+ "workspace": "demo", "q": "fallback retention",
+ "valid_at": 2_500.0, "known_at": 150.0,
+ },
+ )
+ assert fallback.status_code == 200
+ assert fallback.json()["mode"] == "keyword"
+ assert [item["id"] for item in fallback.json()["memories"]] == [old["id"]]
+
+ compact_fallback = client.get(
+ "/api/recall",
+ params={
+ "workspace": "demo", "q": "fallback retention", "response_mode": "compact",
+ "token_budget": 0,
+ },
+ )
+ payload = compact_fallback.json()
+ assert compact_fallback.status_code == 200
+ assert payload["mode"] == "keyword"
+ assert payload["response_mode"] == "compact"
+ assert payload["usage"]["budget_tokens"] == 0
+ assert payload["usage"]["context_tokens"] == 0
+ assert payload["memories"] and "content" not in payload["memories"][0]
+
+
+def test_keyword_recall_fallback_excludes_untrusted_memories(monkeypatch, tmp_path):
+ """A degraded HTTP recall must enforce the same prompt eligibility boundary."""
+ with _client(monkeypatch, tmp_path) as client:
+ svc = v2_api.service()
+ workspace_id = svc.store.get_or_create_workspace("demo")
+ trusted = {"id": svc.engine.remember(
+ "Fallback visibility trusted candidate.",
+ workspace_id=workspace_id, scope=Scope.WORKSPACE,
+ )}
+ untrusted = svc.remember(
+ "Fallback visibility untrusted candidate.",
+ workspace="demo",
+ source="sync",
+ trusted=False,
+ )
+
+ def incompatible_embedder(*_args, **_kwargs):
+ raise ValueError("shapes (256,) and (384,) not aligned")
+
+ monkeypatch.setattr(svc, "recall", incompatible_embedder)
+ response = client.get(
+ "/api/recall",
+ params={"workspace": "demo", "q": "fallback visibility candidate", "k": 1},
+ )
+
+ payload = response.json()
+ assert response.status_code == 200
+ assert payload["mode"] == "keyword"
+ assert [memory["id"] for memory in payload["memories"]] == [trusted["id"]]
+ assert untrusted["id"] not in {memory["id"] for memory in payload["memories"]}
+ assert "untrusted candidate" not in repr(payload)
+
+
+def test_http_memory_api_rejects_backdated_agent_claim_supersession(
+ monkeypatch, tmp_path
+):
+ with _client(monkeypatch, tmp_path) as client:
+ original = client.post(
+ "/api/remember",
+ json={
+ "workspace": "demo",
+ "content": "The deployment window is Friday afternoon.",
+ "valid_from": 2_000.0,
+ },
+ ).json()
+ service = v2_api.service()
+ count_before = len(service.store.list_memories(include_invalid=True))
+ rejected = client.post(
+ "/api/remember",
+ json={
+ "workspace": "demo",
+ "content": "The deployment window is Thursday afternoon.",
+ "valid_from": 1_000.0,
+ },
+ )
+
+ assert rejected.status_code == 400
+ assert service.store.get_memory(original["id"]).valid_to is None
+ assert len(service.store.list_memories(include_invalid=True)) == count_before
+
+
+def test_manual_consolidation_stays_local_but_dreaming_is_cloud_only(
+ monkeypatch, tmp_path
+):
+ with _client(monkeypatch, tmp_path) as client:
+ manual = client.post(
+ "/api/consolidate",
+ json={"workspace": "demo", "dry_run": True, "infer": False},
+ )
+ assert manual.status_code == 200
+ dream = client.post(
+ "/api/consolidate",
+ json={"workspace": "demo", "dry_run": True, "infer": True},
+ )
+ assert dream.status_code == 501
+ assert dream.json()["detail"]["cloud_only"] is True
+
+
+def test_analytics_route_delegates_to_managed_compute(monkeypatch, tmp_path):
+ monkeypatch.setattr(
+ "engraphis.cloud_features.run_managed_job",
+ lambda service, workspace, kind: {
+ "result": {
+ "kind": kind,
+ "generation": 4,
+ "totals": {"live": 1},
+ }
+ },
+ )
+ with _client(monkeypatch, tmp_path) as client:
+ response = client.get("/api/analytics?workspace=demo")
+ assert response.status_code == 200
+ assert response.json()["kind"] == "analytics"
+ assert response.json()["generation"] == 4
+
+
+def test_unconnected_automation_returns_a_structured_auth_error(monkeypatch, tmp_path):
+ for name in (
+ "ENGRAPHIS_CLOUD_ACCESS_TOKEN",
+ "ENGRAPHIS_CLOUD_ORGANIZATION_ID",
+ "ENGRAPHIS_CLOUD_COMPUTE_URL",
+ "ENGRAPHIS_CLOUD_REFRESH_CREDENTIAL",
+ "ENGRAPHIS_CLOUD_CONTROL_URL",
+ ):
+ monkeypatch.delenv(name, raising=False)
+ monkeypatch.setenv("ENGRAPHIS_STATE_DIR", str(tmp_path / "unconnected-state"))
+
+ with _client(monkeypatch, tmp_path) as client:
+ response = client.get("/api/automation?workspace=demo")
+
+ assert response.status_code == 401
+ # The copy is ``_public_session_error(401)``: fixed, status-keyed, and actionable. The
+ # generic placeholder told an unconnected customer nothing they could act on.
+ assert response.json()["detail"] == {
+ "error": "Connect this installation to Engraphis Cloud to use hosted features.",
+ "managed_cloud": True,
+ "transient": False,
+ "code": "cloud_unconfigured",
+ }
+
+
+def test_hosted_automation_accepts_the_cloud_policy_field(monkeypatch, tmp_path):
+ saved = {}
+
+ class _Cloud:
+ def upload_snapshot(self, workspace_id, snapshot):
+ return {"generation": snapshot["generation"]}
+
+ def get_policy(self, workspace_id):
+ return {"enabled": False, "cadence_minutes": 1440, "dream_enabled": False}
+
+ def save_policy(self, workspace_id, policy):
+ saved.update(policy)
+ return {"version": 2}
+
+ monkeypatch.setattr(
+ "engraphis.cloud_features.build_managed_snapshot",
+ lambda service, workspace: ("ws_cloud", {"generation": 1}),
+ )
+ monkeypatch.setattr(
+ "engraphis.cloud_features.CloudFeatureClient.from_environment",
+ lambda workspace_id=None: _Cloud(),
+ )
+ with _client(monkeypatch, tmp_path) as client:
+ response = client.post(
+ "/api/automation",
+ json={"enabled": True, "dream_enabled": True, "cadence_hours": 12},
+ )
+ assert response.status_code == 200
+ assert response.json()["dream_enabled"] is True
+ assert saved["dream_enabled"] is True
+
+
+def test_first_hosted_automation_view_bootstraps_the_recommended_policy(
+ monkeypatch, tmp_path
+):
+ """A connected Pro/Team workspace starts maintaining itself without a toggle."""
+
+ uploaded = []
+ saved = []
+
+ class _Cloud:
+ organization_id = "org_test"
+
+ def get_policy(self, workspace_id):
+ # Version zero is the private Cloud's documented no-policy sentinel.
+ return {"enabled": False, "cadence_minutes": 1440, "version": 0}
+
+ def upload_snapshot(self, workspace_id, snapshot):
+ uploaded.append((workspace_id, snapshot))
+ return {"generation": snapshot["generation"]}
+
+ def save_policy(self, workspace_id, policy):
+ saved.append((workspace_id, policy))
+ return {"version": 1}
+
+ def list_jobs(self, workspace_id, *, limit=10):
+ return {"jobs": []}
+
+ monkeypatch.setattr(
+ "engraphis.cloud_features.build_managed_snapshot",
+ lambda service, workspace: ("ws_cloud", {"generation": 7}),
+ )
+ monkeypatch.setattr(
+ "engraphis.cloud_features.CloudFeatureClient.from_environment",
+ lambda workspace_id=None: _Cloud(),
+ )
+ with _client(monkeypatch, tmp_path) as client:
+ response = client.get("/api/automation")
+
+ assert response.status_code == 200
+ assert response.json()["enabled"] is True
+ assert response.json()["dream"] is True
+ assert uploaded == [("ws_cloud", {"generation": 7})]
+ assert saved == [("ws_cloud", {
+ "enabled": True,
+ "cadence_minutes": 1440,
+ "dream_enabled": True,
+ "dream_min_new": 25,
+ "dream_idle_minutes": 15,
+ "infer": False,
+ })]
+
+
+def test_first_automation_policy_retry_does_not_upload_the_snapshot_twice(
+ monkeypatch, tmp_path
+):
+ """A failed policy write resumes after the already successful private upload."""
+
+ from engraphis.cloud_features import CloudFeatureError
+
+ uploaded = []
+ saved = []
+ builds = []
+
+ class _Cloud:
+ organization_id = "org_test"
+
+ def get_policy(self, workspace_id):
+ return {"enabled": False, "cadence_minutes": 1440, "version": 0}
+
+ def upload_snapshot(self, workspace_id, snapshot):
+ uploaded.append((workspace_id, snapshot))
+ return {"generation": snapshot["generation"]}
+
+ def save_policy(self, workspace_id, policy):
+ saved.append((workspace_id, policy))
+ if len(saved) == 1:
+ raise CloudFeatureError(
+ "Engraphis Cloud is temporarily unavailable.",
+ status=503,
+ transient=True,
+ )
+ return {"version": 1}
+
+ def list_jobs(self, workspace_id, *, limit=10):
+ return {"jobs": []}
+
+ def _snapshot(service, workspace):
+ builds.append(workspace)
+ return "ws_cloud", {"generation": 7}
+
+ monkeypatch.setattr("engraphis.cloud_features.build_managed_snapshot", _snapshot)
+ monkeypatch.setattr(
+ "engraphis.cloud_features.CloudFeatureClient.from_environment",
+ lambda workspace_id=None: _Cloud(),
+ )
+ with _client(monkeypatch, tmp_path) as client:
+ first = client.get("/api/automation")
+ second = client.get("/api/automation")
+
+ assert first.status_code == 503
+ assert second.status_code == 200
+ assert len(builds) == 1
+ assert uploaded == [("ws_cloud", {"generation": 7})]
+ assert len(saved) == 2
+
+
+def test_concurrent_first_automation_views_upload_one_snapshot(monkeypatch, tmp_path):
+ """Parallel dashboard reads serialize the sensitive first-bootstrap upload."""
+
+ uploaded = []
+ saved = []
+ started = threading.Event()
+ release_upload = threading.Event()
+
+ class _Cloud:
+ organization_id = "org_concurrent"
+
+ def get_policy(self, workspace_id):
+ return {"enabled": False, "cadence_minutes": 1440, "version": 0}
+
+ def upload_snapshot(self, workspace_id, snapshot):
+ uploaded.append((workspace_id, snapshot))
+ started.set()
+ assert release_upload.wait(timeout=5)
+ return {"generation": snapshot["generation"]}
+
+ def save_policy(self, workspace_id, policy):
+ saved.append((workspace_id, policy))
+ return {"version": 1}
+
+ def list_jobs(self, workspace_id, *, limit=10):
+ return {"jobs": []}
+
+ monkeypatch.setattr(
+ "engraphis.cloud_features.build_managed_snapshot",
+ lambda service, workspace: ("ws_cloud", {"generation": 7}),
+ )
+ monkeypatch.setattr(
+ "engraphis.cloud_features.CloudFeatureClient.from_environment",
+ lambda workspace_id=None: _Cloud(),
+ )
+ with _client(monkeypatch, tmp_path):
+ with ThreadPoolExecutor(max_workers=2) as pool:
+ first = pool.submit(v2_api.automation_get)
+ assert started.wait(timeout=5)
+ second = pool.submit(v2_api.automation_get)
+ release_upload.set()
+ assert first.result(timeout=5)["enabled"] is True
+ follower = second.result(timeout=5)
+ assert follower["enabled"] is True
+ assert follower["version"] == 1
+
+ assert uploaded == [("ws_cloud", {"generation": 7})]
+ assert len(saved) == 1
+
+
+def test_reading_or_disabling_automation_never_uploads_memory_content(
+ monkeypatch, tmp_path
+):
+ saved = {}
+
+ class _Cloud:
+ def get_policy(self, workspace_id):
+ return {"enabled": True, "cadence_minutes": 60, "dream_enabled": True}
+
+ def list_jobs(self, workspace_id, *, limit=10):
+ return {"jobs": []}
+
+ def save_policy(self, workspace_id, policy):
+ saved.update(policy)
+ return {"version": 3}
+
+ def _unexpected_upload(*args, **kwargs):
+ raise AssertionError("policy inspection must not build or upload a snapshot")
+
+ monkeypatch.setattr(
+ "engraphis.cloud_features.build_managed_snapshot",
+ _unexpected_upload,
+ )
+ monkeypatch.setattr(
+ "engraphis.cloud_features.CloudFeatureClient.from_environment",
+ lambda workspace_id=None: _Cloud(),
+ )
+ with _client(monkeypatch, tmp_path) as client:
+ assert client.get("/api/automation").status_code == 200
+ response = client.post("/api/automation", json={"enabled": False})
+ assert response.status_code == 200
+ assert saved["enabled"] is False
+
+
+def test_automation_and_maintenance_use_the_selected_workspace(monkeypatch, tmp_path):
+ policy_workspaces = []
+ snapshot_workspaces = []
+ maintenance_workspaces = []
+
+ class _Cloud:
+ def get_policy(self, workspace_id):
+ policy_workspaces.append(workspace_id)
+ return {"enabled": False, "cadence_minutes": 60, "dream_enabled": True}
+
+ def list_jobs(self, workspace_id, *, limit=10):
+ policy_workspaces.append(workspace_id)
+ return {"jobs": []}
+
+ def upload_snapshot(self, workspace_id, snapshot):
+ snapshot_workspaces.append(workspace_id)
+ return {"generation": snapshot["generation"]}
+
+ def save_policy(self, workspace_id, policy):
+ policy_workspaces.append(workspace_id)
+ return {"version": 1}
+
+ def snapshot(service, workspace):
+ snapshot_workspaces.append(workspace)
+ return service._lookup_workspace(workspace), {"generation": 1}
+
+ def managed_job(service, workspace, kind):
+ maintenance_workspaces.append((workspace, kind))
+ return {"result": {"kind": kind}}
+
+ monkeypatch.setattr("engraphis.cloud_features.build_managed_snapshot", snapshot)
+ monkeypatch.setattr("engraphis.cloud_features.run_managed_job", managed_job)
+ monkeypatch.setattr(
+ "engraphis.cloud_features.CloudFeatureClient.from_environment",
+ lambda workspace_id=None: _Cloud(),
+ )
+ with _client(monkeypatch, tmp_path) as client:
+ beta_id = client.app.state.service._lookup_workspace("beta")
+ demo_id = client.app.state.service._lookup_workspace("demo")
+ assert client.get("/api/automation?workspace=beta").status_code == 200
+ assert client.post(
+ "/api/automation?workspace=beta", json={"enabled": True}
+ ).status_code == 200
+ assert client.post(
+ "/api/maintenance/run?workspace=beta", json={"dry_run": True}
+ ).status_code == 200
+
+ assert beta_id in policy_workspaces
+ assert demo_id not in policy_workspaces
+ assert "beta" in snapshot_workspaces
+ assert maintenance_workspaces == [("beta", "consolidate")]
+
+
+def test_automation_workspace_query_unknown_is_not_replaced_by_legacy_default(
+ monkeypatch, tmp_path
+):
+ with _client(monkeypatch, tmp_path) as client:
+ for method, path, payload in (
+ (client.get, "/api/automation?workspace=missing", None),
+ (client.post, "/api/automation?workspace=missing", {"enabled": False}),
+ (client.post, "/api/maintenance/run?workspace=missing", {"dry_run": True}),
+ ):
+ response = method(path, json=payload) if payload is not None else method(path)
+ assert response.status_code == 404
+
+
+def test_dashboard_automation_uses_active_workspace_and_discloses_upload_boundary():
+ source = Path(__file__).parents[1] / "engraphis" / "static" / "dashboard.js"
+ source = source.read_text(encoding="utf-8")
+ assert "/automation?workspace=" in source
+ assert "/maintenance/run?workspace=" in source
+ assert "Preview snapshot" not in source
+ assert "uploads the selected workspace’s normal and sensitive memory content" in source
+ # The upload boundary is still disclosed, but consent now travels with the cloud
+ # account: the dashboard must not name the operator override anywhere.
+ assert "ENGRAPHIS_MANAGED_COMPUTE_CONSENT" not in source
+ assert "Hosted work is automatic with Pro." in source
+
+
+def test_portfolio_and_report_analytics_are_hosted_only(monkeypatch, tmp_path):
+ with _client(monkeypatch, tmp_path) as client:
+ assert client.get("/api/analytics/portfolio").status_code == 501
+ assert client.get("/api/analytics/export?workspace=demo").status_code == 501
+
+
+def test_raw_owner_export_is_free_and_signed_export_is_honestly_unimplemented(
+ monkeypatch, tmp_path
+):
+ """The signed variant must not claim to exist somewhere else.
+
+ It previously answered ``cloud_only: True`` — but Engraphis Cloud has no export route,
+ no supported hosted export capability, so that pointed a customer at a
+ product that does not exist. The 501 now says the capability is unimplemented and names
+ the working unsigned export instead.
+ """
+
+ with _client(monkeypatch, tmp_path) as client:
+ raw = client.get("/api/export?workspace=demo")
+ assert raw.status_code == 200
+ assert raw.json()["counts"]["memories"] >= 1
+ signed = client.get("/api/export?workspace=demo&signed=true")
+ assert signed.status_code == 501
+ detail = signed.json()["detail"]
+ assert detail["implemented"] is False
+ assert detail["alternative"] == "/export"
+ assert "cloud_only" not in detail
+ assert "Engraphis Cloud" not in detail["error"]
+
+
+def test_health_and_readiness_remain_public(monkeypatch, tmp_path):
+ with _client(monkeypatch, tmp_path) as client:
+ assert client.get("/api/health").status_code == 200
+ assert client.get("/api/ready").status_code == 200
+
+
+def test_dashboard_exception_responses_do_not_echo_untrusted_exception_text():
+ secret = "https://provider.example/?api_key=do-not-return-this"
+
+ def fail_with(exc):
+ raise exc
+
+ with pytest.raises(HTTPException) as internal:
+ v2_api._run(fail_with, RuntimeError(secret))
+ assert internal.value.status_code == 500
+ assert internal.value.detail == {"error": "internal server error"}
+ assert secret not in repr(internal.value.detail)
+
+ with pytest.raises(HTTPException) as validation:
+ v2_api._run(fail_with, ValidationError(secret))
+ assert validation.value.status_code == 400
+ assert validation.value.detail == {"error": "invalid request"}
+ assert secret not in repr(validation.value.detail)
+
+ with pytest.raises(HTTPException) as downstream:
+ v2_api._run(fail_with, HTTPException(status_code=418, detail={"error": secret}))
+ assert downstream.value.status_code == 418
+ assert downstream.value.detail == {"error": "request rejected"}
+ assert secret not in repr(downstream.value.detail)
+
+ with pytest.raises(HTTPException) as invalid_status:
+ v2_api._run(fail_with, HTTPException(status_code=999, detail={"error": secret}))
+ assert invalid_status.value.status_code == 500
+ assert invalid_status.value.detail == {"error": "internal server error"}
+ assert secret not in repr(invalid_status.value.detail)
+
+ with pytest.raises(HTTPException) as mismatch:
+ v2_api._run(fail_with, ValueError(f"{secret}: shapes 256 and 384 are not aligned"))
+ assert mismatch.value.status_code == 409
+ assert mismatch.value.detail["embedder"] is True
+ assert secret not in repr(mismatch.value.detail)
+
+ with pytest.raises(HTTPException) as ordinary_value_error:
+ v2_api._run(fail_with, ValueError(secret))
+ assert ordinary_value_error.value.status_code == 400
+ assert ordinary_value_error.value.detail == {"error": "invalid request"}
+ assert secret not in repr(ordinary_value_error.value.detail)
+
+
+def test_dashboard_engine_value_error_is_a_sanitized_client_error(monkeypatch, tmp_path):
+ secret = "malformed document details must stay private"
+ with _client(monkeypatch, tmp_path) as client:
+ def reject_document(*_args, **_kwargs):
+ raise ValueError(secret)
+
+ monkeypatch.setattr(client.app.state.service, "remember", reject_document)
+ response = client.post(
+ "/api/remember",
+ json={"content": "client document", "workspace": "demo"},
+ )
+
+ assert response.status_code == 400
+ assert response.json() == {"detail": {"error": "invalid request"}}
+ assert secret not in response.text
+
+
+def test_managed_cloud_errors_forward_only_bounded_public_copy():
+ """``_managed_call`` forwards the message; the bound is the boundary's own check.
+
+ ``CloudFeatureError`` is the already-redacted form -- every raise site builds it from
+ fixed, status-keyed copy -- so its text is what the customer should read. The bound
+ here is not the redaction, it is the guard for a message that is *not* that fixed copy:
+ anything oversized, empty, or carrying control characters is dropped for the generic
+ placeholder rather than rendered into a JSON error body.
+ """
+
+ def fail_with(exc):
+ raise exc
+
+ for message in ("x" * 301, "", "connection\x00reset", "trace\x1b[31m"):
+ with pytest.raises(HTTPException) as caught:
+ v2_api._managed_call(fail_with, CloudFeatureError(message, status=502))
+ assert caught.value.status_code == 502
+ assert caught.value.detail == {
+ "error": v2_api._MANAGED_ERROR_FALLBACK, "managed_cloud": True,
+ "transient": False,
+ }
+
+ with pytest.raises(HTTPException) as consent:
+ v2_api._managed_call(
+ fail_with,
+ CloudFeatureError(
+ "Managed compute is turned off for this installation.",
+ status=409, code="consent_required",
+ ),
+ )
+ assert consent.value.status_code == 409
+ assert consent.value.detail == {
+ "error": "Managed compute is turned off for this installation.",
+ "managed_cloud": True,
+ "transient": False,
+ "code": "consent_required",
+ }
+
+ with pytest.raises(HTTPException) as unconfigured:
+ v2_api._managed_call(
+ fail_with,
+ CloudFeatureError(
+ "Connect this installation to Engraphis Cloud to use hosted features.",
+ status=401, code="cloud_unconfigured",
+ ),
+ )
+ assert unconfigured.value.status_code == 401
+ assert unconfigured.value.detail == {
+ "error": "Connect this installation to Engraphis Cloud to use hosted features.",
+ "managed_cloud": True,
+ "transient": False,
+ "code": "cloud_unconfigured",
+ }
+
+
+@pytest.mark.parametrize("status", (401, 402, 403))
+def test_managed_authorization_denial_settles_local_entitlement(monkeypatch, status):
+ """A live hosted denial must immediately retire stale paid presentation state."""
+
+ calls = []
+ monkeypatch.setattr(v2_api, "_record_authoritative_denial", lambda: calls.append(status))
+
+ def fail_with(exc):
+ raise exc
+
+ with pytest.raises(HTTPException) as caught:
+ v2_api._managed_call(
+ fail_with, CloudFeatureError("Engraphis Cloud authorization was rejected.",
+ status=status),
+ )
+
+ assert caught.value.status_code == status
+ assert calls == [status]
+
+
+@pytest.mark.parametrize("status", (409, 429, 503))
+def test_managed_non_authorization_failures_do_not_settle_entitlement(monkeypatch, status):
+ """Conflicts and outages do not prove that a subscription or membership changed."""
+
+ calls = []
+ monkeypatch.setattr(v2_api, "_record_authoritative_denial", lambda: calls.append(status))
+
+ def fail_with(exc):
+ raise exc
+
+ with pytest.raises(HTTPException):
+ v2_api._managed_call(
+ fail_with, CloudFeatureError("Engraphis Cloud temporarily failed.", status=status),
+ )
+
+ assert calls == []
+
+
+def _managed_http_failure(monkeypatch, status: int) -> HTTPException:
+ """Drive one real hosted request against a control plane that answers ``status``."""
+
+ class _Opener:
+ def open(self, request, timeout=None):
+ raise urllib.error.HTTPError(
+ "https://compute.example.test/private", status, "failure", {},
+ io.BytesIO(b'{"detail": "provider-internals https://backend.invalid"}'),
+ )
+
+ monkeypatch.setattr(
+ cloud_features, "build_pinned_https_opener", lambda *handlers: _Opener()
+ )
+ client = cloud_features.CloudFeatureClient(
+ "https://compute.example.test", "org_1", "token"
+ )
+ with pytest.raises(HTTPException) as caught:
+ v2_api._managed_call(client._request, "GET", "/private")
+ return caught.value
+
+
+def test_a_managed_outage_is_distinguishable_from_a_workspace_conflict(monkeypatch):
+ """The defect: every hosted failure rendered as one fixed, unactionable string.
+
+ ``cloud_features._public_http_error`` already produces redacted, status-keyed copy that
+ tells a retryable outage apart from a conflict the customer has to fix -- and
+ ``_managed_call`` threw all of it away, so the dashboard's error branch could only ever
+ show "managed cloud operation failed" for a 429, a 5xx and a 409 alike.
+ """
+
+ busy = _managed_http_failure(monkeypatch, 429)
+ down = _managed_http_failure(monkeypatch, 503)
+ conflict = _managed_http_failure(monkeypatch, 409)
+
+ assert busy.status_code == 429
+ assert busy.detail["transient"] is True
+ assert "temporarily busy" in busy.detail["error"], busy.detail["error"]
+
+ assert down.status_code == 503
+ assert down.detail["transient"] is True
+ assert "temporarily unavailable" in down.detail["error"], down.detail["error"]
+
+ assert conflict.status_code == 409
+ assert conflict.detail["transient"] is False
+ assert "workspace state" in conflict.detail["error"], conflict.detail["error"]
+
+ messages = {busy.detail["error"], down.detail["error"], conflict.detail["error"]}
+ assert len(messages) == 3, "the dashboard still cannot tell these three apart"
+ assert v2_api._MANAGED_ERROR_FALLBACK not in messages
+ # Forwarding the public copy must not forward the provider's body with it.
+ assert all("provider-internals" not in text for text in messages)
+ assert all("backend.invalid" not in text for text in messages)
+
+
+def test_every_managed_cloud_error_message_is_fixed_local_copy():
+ """The invariant that makes forwarding safe, pinned against future raise sites.
+
+ ``_managed_call`` may forward a ``CloudFeatureError`` message only because every one of
+ them is built from a literal in this repository -- never from a provider body, a
+ ``CloudSessionError``, or a local path. A raise site that interpolated a runtime value
+ would silently turn this boundary into a reflection point, so the shape is asserted
+ rather than trusted.
+
+ Three forms are accepted: a string literal; a name bound from ``_public_http_error`` /
+ ``_public_session_error`` (both of which switch on a bare integer status and return
+ fixed copy); and the one audited ``%`` template, below.
+ """
+
+ source = Path(cloud_features.__file__).read_text(encoding="utf-8")
+ tree = ast.parse(source)
+
+ public_copy = {"_public_http_error", "_public_session_error"}
+ from_public_copy = set()
+ for node in ast.walk(tree):
+ if not isinstance(node, ast.Assign) or not isinstance(node.value, ast.Call):
+ continue
+ called = node.value.func
+ if not isinstance(called, ast.Name) or called.id not in public_copy:
+ continue
+ for target in node.targets:
+ elements = target.elts if isinstance(target, ast.Tuple) else [target]
+ from_public_copy.update(
+ item.id for item in elements if isinstance(item, ast.Name)
+ )
+ assert from_public_copy, "the fixed-copy helpers are no longer bound to a name"
+
+ interpolated = []
+ for node in ast.walk(tree):
+ if not isinstance(node, ast.Call):
+ continue
+ name = node.func.id if isinstance(node.func, ast.Name) else None
+ if name != "CloudFeatureError" or not node.args:
+ continue
+ message = node.args[0]
+ if isinstance(message, ast.Constant) and isinstance(message.value, str):
+ continue
+ if isinstance(message, ast.Name) and message.id in from_public_copy:
+ continue
+ # ``"literal %s" % (...)`` is allowed only where the substituted values are
+ # themselves constrained to local literals; ``run_job`` is the single such site
+ # and its ``status`` is guarded by an ``in {"failed", "canceled"}`` membership
+ # test one line above. Anything else -- an f-string, a bare name, a concatenated
+ # response field -- is a reflection risk and fails here.
+ if (isinstance(message, ast.BinOp) and isinstance(message.op, ast.Mod)
+ and isinstance(message.left, ast.Constant)
+ and message.left.value == "Managed %s did not complete (%s)."):
+ continue
+ interpolated.append((node.lineno, ast.dump(message)[:120]))
+
+ assert interpolated == [], (
+ "a CloudFeatureError message is no longer fixed local copy; _managed_call "
+ "forwards it to the customer: %r" % (interpolated,)
+ )
diff --git a/tests/test_savings.py b/tests/test_savings.py
index 713394b6..04cfaebf 100644
--- a/tests/test_savings.py
+++ b/tests/test_savings.py
@@ -1,218 +1,218 @@
-import pytest
-
-from engraphis import __version__
-from engraphis.core.savings import SavingsEstimate, annotate_usage, estimate_savings
-from engraphis.core.store import Store
-from engraphis.service import MemoryService, ValidationError
-
-
-@pytest.mark.parametrize(
- ("operation", "intent", "adaptive_mode", "basis", "confidence", "eligible"),
- [
- ("adaptive_context", None, "retrieval", "history_retrieval", "high", True),
- ("adaptive_context", None, "history_fallback", "history_fallback", "medium", True),
- ("adaptive_context", None, "history_bypass", "history_bypass", "none", False),
- (
- "adaptive_context",
- None,
- "low_confidence_abstain",
- "low_confidence_abstain",
- "none",
- False,
- ),
- ("recall", "recall_context", None, "packed_context", "medium", True),
- ("grounded_recall", None, None, "packed_context", "medium", True),
- ("proactive_context", None, None, "packed_context", "medium", True),
- ("recall", "recall", None, "unclassified", "unknown", False),
- ],
-)
-def test_estimator_classifies_each_delivery_basis(
- operation, intent, adaptive_mode, basis, confidence, eligible
-):
- estimate = estimate_savings(
- operation=operation,
- intent=intent,
- adaptive_mode=adaptive_mode,
- baseline_tokens=100,
- emitted_tokens=40,
- token_counter="engraphis.regex.v1",
- release_version="1.5.0",
- )
-
- assert isinstance(estimate, SavingsEstimate)
- assert estimate.basis == basis
- assert estimate.confidence == confidence
- assert estimate.eligible is eligible
- assert estimate.saved_tokens == (60 if eligible else 0)
- assert 0 <= estimate.saved_tokens <= estimate.baseline_tokens
- assert 0 <= estimate.savings_ratio <= 1
- assert estimate.release_version == "1.5.0"
-
-
-def test_estimator_is_conservative_for_bad_counts_and_annotates_existing_usage():
- usage = annotate_usage(
- {"source_tokens": 90, "context_tokens": 30, "saved_tokens": 60,
- "token_counter": "engraphis.regex.v1"},
- operation="adaptive_context",
- adaptive_mode="history_fallback",
- baseline_tokens=90,
- emitted_tokens=30,
- release_version=__version__,
- )
-
- assert usage["estimated_saved_tokens"] == 60
- assert usage["savings_eligible"] is True
- assert usage["release_version"] == "1.5.0"
- abstained = estimate_savings(
- operation="adaptive_context",
- adaptive_mode="low_confidence_abstain",
- baseline_tokens=float("nan"),
- emitted_tokens=0,
- )
- assert abstained.saved_tokens == 0
- assert abstained.baseline_tokens == 0
-
-
-def _usage(baseline, emitted, *, counter, release="1.5.0", eligible=True,
- basis="history_retrieval", confidence="high"):
- saved = max(0, baseline - emitted) if eligible else 0
- return {
- "source_tokens": baseline,
- "context_tokens": emitted,
- "saved_tokens": saved,
- "budget_tokens": baseline,
- "packed_count": 1,
- "omitted_count": 0,
- "token_counter": counter,
- "baseline_tokens": baseline,
- "emitted_tokens": emitted,
- "estimated_saved_tokens": saved,
- "estimated_savings_ratio": saved / baseline if baseline else 0.0,
- "savings_basis": basis,
- "savings_confidence": confidence,
- "savings_eligible": eligible,
- "release_version": release,
- }
-
-
-def test_context_savings_aggregates_estimates_filters_releases_and_counters():
- store = Store(":memory:")
- wid = store.get_or_create_workspace("savings")
- rid = store.get_or_create_repo(wid, "repo")
- first = store.record_receipt(
- "adaptive_context",
- workspace_id=wid,
- repo_id=rid,
- metadata={"adaptive_mode": "retrieval", "token_usage": _usage(
- 100, 40, counter="engraphis.regex.v1"
- )},
- )
- second = store.record_receipt(
- "adaptive_context",
- workspace_id=wid,
- repo_id=rid,
- metadata={"adaptive_mode": "history_bypass", "token_usage": _usage(
- 80, 80, counter="engraphis.regex.v1", eligible=False,
- basis="history_bypass", confidence="none"
- )},
- )
- third = store.record_receipt(
- "recall",
- workspace_id=wid,
- repo_id=rid,
- metadata={"intent": "recall_context", "token_usage": _usage(
- 50, 20, counter="estimate_tokens", release="1.4.0",
- basis="packed_context", confidence="medium"
- )},
- )
- old = store.record_receipt(
- "recall",
- workspace_id=wid,
- repo_id=rid,
- metadata={"intent": "recall_context", "token_usage": {
- "source_tokens": 20, "context_tokens": 10, "saved_tokens": 10,
- "token_counter": "engraphis.regex.v1",
- }},
- )
- for timestamp, receipt in ((100.0, first), (110.0, second), (120.0, third), (130.0, old)):
- store.conn.execute(
- "UPDATE operation_receipts SET ts=? WHERE id=?", (timestamp, receipt["id"])
- )
- store.conn.commit()
-
- summary = store.context_savings(
- workspace_id=wid, repo_id=rid, from_ts=99, to_ts=121
- )
- assert summary["estimated"]["eligible_receipt_count"] == 2
- assert summary["estimated"]["excluded_receipt_count"] == 1
- assert summary["estimated"]["unclassified_receipt_count"] == 0
- assert summary["estimated"]["baseline_tokens"] == 150
- assert summary["estimated"]["emitted_tokens"] == 60
- assert summary["estimated"]["saved_tokens"] == 90
- assert {row["token_counter"] for row in summary["estimated"]["by_token_counter"]} == {
- "engraphis.regex.v1", "estimate_tokens"
- }
- assert summary["period"] == {"from_ts": 99, "to_ts": 121}
- all_time = store.context_savings(workspace_id=wid, repo_id=rid)
- assert all_time["estimated"]["unclassified_receipt_count"] == 1
-
- current = store.context_savings(
- workspace_id=wid, repo_id=rid, release_version="1.5.0"
- )
- assert current["receipt_count"] == 2
- assert current["usage_receipt_count"] == 2
- assert current["estimated"]["eligible_receipt_count"] == 1
- assert current["estimated"]["saved_tokens"] == 60
- assert current["estimated"]["by_basis"][0]["basis"] == "history_retrieval"
-
- with pytest.raises(ValueError, match="semantic version"):
- store.context_savings(workspace_id=wid, release_version="not-a-release")
-
-
-def test_service_context_savings_filters_and_new_receipts_are_versioned():
- service = MemoryService.create(":memory:", graph_extractor="none")
- service.remember("Versioned context delivery.", workspace="versioned", scope="workspace")
- service.recall(
- "context delivery",
- workspace="versioned",
- token_budget=32,
- response_mode="compact",
- intent="recall_context",
- )
- receipt = service.receipt_log(workspace="versioned")["entries"][0]
- usage = receipt["metadata"]["token_usage"]
- assert usage["release_version"] == __version__
- assert usage["savings_basis"] == "packed_context"
- assert usage["savings_eligible"] is True
- filtered = service.context_savings(
- workspace="versioned", release_version=__version__,
- from_ts=0, to_ts=9_999_999_999,
- )
- assert filtered["estimated"]["eligible_receipt_count"] == 1
- with pytest.raises(ValidationError, match="semantic version"):
- service.context_savings(workspace="versioned", release_version="legacy")
-
-
-def test_context_savings_ignores_gateway_copies_and_rejects_noncanonical_estimates():
- store = Store(":memory:")
- wid = store.get_or_create_workspace("gateway-savings")
- authoritative = _usage(100, 40, counter="engraphis.regex.v1")
- store.record_receipt(
- "adaptive_context", workspace_id=wid,
- metadata={"token_usage": authoritative},
- )
- store.record_receipt(
- "smart_gateway", workspace_id=wid,
- metadata={"token_usage": authoritative},
- )
- noncanonical = _usage(80, 20, counter="engraphis.regex.v1")
- noncanonical["estimated_savings_ratio"] = 0.1
- store.record_receipt(
- "adaptive_context", workspace_id=wid,
- metadata={"token_usage": noncanonical},
- )
-
- summary = store.context_savings(workspace_id=wid)
- assert summary["estimated"]["eligible_receipt_count"] == 1
- assert summary["estimated"]["saved_tokens"] == 60
- assert summary["estimated"]["invalid_estimate_count"] == 1
+import pytest
+
+from engraphis import __version__
+from engraphis.core.savings import SavingsEstimate, annotate_usage, estimate_savings
+from engraphis.core.store import Store
+from engraphis.service import MemoryService, ValidationError
+
+
+@pytest.mark.parametrize(
+ ("operation", "intent", "adaptive_mode", "basis", "confidence", "eligible"),
+ [
+ ("adaptive_context", None, "retrieval", "history_retrieval", "high", True),
+ ("adaptive_context", None, "history_fallback", "history_fallback", "medium", True),
+ ("adaptive_context", None, "history_bypass", "history_bypass", "none", False),
+ (
+ "adaptive_context",
+ None,
+ "low_confidence_abstain",
+ "low_confidence_abstain",
+ "none",
+ False,
+ ),
+ ("recall", "recall_context", None, "packed_context", "medium", True),
+ ("grounded_recall", None, None, "packed_context", "medium", True),
+ ("proactive_context", None, None, "packed_context", "medium", True),
+ ("recall", "recall", None, "unclassified", "unknown", False),
+ ],
+)
+def test_estimator_classifies_each_delivery_basis(
+ operation, intent, adaptive_mode, basis, confidence, eligible
+):
+ estimate = estimate_savings(
+ operation=operation,
+ intent=intent,
+ adaptive_mode=adaptive_mode,
+ baseline_tokens=100,
+ emitted_tokens=40,
+ token_counter="engraphis.regex.v1",
+ release_version="1.5",
+ )
+
+ assert isinstance(estimate, SavingsEstimate)
+ assert estimate.basis == basis
+ assert estimate.confidence == confidence
+ assert estimate.eligible is eligible
+ assert estimate.saved_tokens == (60 if eligible else 0)
+ assert 0 <= estimate.saved_tokens <= estimate.baseline_tokens
+ assert 0 <= estimate.savings_ratio <= 1
+ assert estimate.release_version == "1.5"
+
+
+def test_estimator_is_conservative_for_bad_counts_and_annotates_existing_usage():
+ usage = annotate_usage(
+ {"source_tokens": 90, "context_tokens": 30, "saved_tokens": 60,
+ "token_counter": "engraphis.regex.v1"},
+ operation="adaptive_context",
+ adaptive_mode="history_fallback",
+ baseline_tokens=90,
+ emitted_tokens=30,
+ release_version=__version__,
+ )
+
+ assert usage["estimated_saved_tokens"] == 60
+ assert usage["savings_eligible"] is True
+ assert usage["release_version"] == "1.5"
+ abstained = estimate_savings(
+ operation="adaptive_context",
+ adaptive_mode="low_confidence_abstain",
+ baseline_tokens=float("nan"),
+ emitted_tokens=0,
+ )
+ assert abstained.saved_tokens == 0
+ assert abstained.baseline_tokens == 0
+
+
+def _usage(baseline, emitted, *, counter, release="1.5", eligible=True,
+ basis="history_retrieval", confidence="high"):
+ saved = max(0, baseline - emitted) if eligible else 0
+ return {
+ "source_tokens": baseline,
+ "context_tokens": emitted,
+ "saved_tokens": saved,
+ "budget_tokens": baseline,
+ "packed_count": 1,
+ "omitted_count": 0,
+ "token_counter": counter,
+ "baseline_tokens": baseline,
+ "emitted_tokens": emitted,
+ "estimated_saved_tokens": saved,
+ "estimated_savings_ratio": saved / baseline if baseline else 0.0,
+ "savings_basis": basis,
+ "savings_confidence": confidence,
+ "savings_eligible": eligible,
+ "release_version": release,
+ }
+
+
+def test_context_savings_aggregates_estimates_filters_releases_and_counters():
+ store = Store(":memory:")
+ wid = store.get_or_create_workspace("savings")
+ rid = store.get_or_create_repo(wid, "repo")
+ first = store.record_receipt(
+ "adaptive_context",
+ workspace_id=wid,
+ repo_id=rid,
+ metadata={"adaptive_mode": "retrieval", "token_usage": _usage(
+ 100, 40, counter="engraphis.regex.v1"
+ )},
+ )
+ second = store.record_receipt(
+ "adaptive_context",
+ workspace_id=wid,
+ repo_id=rid,
+ metadata={"adaptive_mode": "history_bypass", "token_usage": _usage(
+ 80, 80, counter="engraphis.regex.v1", eligible=False,
+ basis="history_bypass", confidence="none"
+ )},
+ )
+ third = store.record_receipt(
+ "recall",
+ workspace_id=wid,
+ repo_id=rid,
+ metadata={"intent": "recall_context", "token_usage": _usage(
+ 50, 20, counter="estimate_tokens", release="1.4.0",
+ basis="packed_context", confidence="medium"
+ )},
+ )
+ old = store.record_receipt(
+ "recall",
+ workspace_id=wid,
+ repo_id=rid,
+ metadata={"intent": "recall_context", "token_usage": {
+ "source_tokens": 20, "context_tokens": 10, "saved_tokens": 10,
+ "token_counter": "engraphis.regex.v1",
+ }},
+ )
+ for timestamp, receipt in ((100.0, first), (110.0, second), (120.0, third), (130.0, old)):
+ store.conn.execute(
+ "UPDATE operation_receipts SET ts=? WHERE id=?", (timestamp, receipt["id"])
+ )
+ store.conn.commit()
+
+ summary = store.context_savings(
+ workspace_id=wid, repo_id=rid, from_ts=99, to_ts=121
+ )
+ assert summary["estimated"]["eligible_receipt_count"] == 2
+ assert summary["estimated"]["excluded_receipt_count"] == 1
+ assert summary["estimated"]["unclassified_receipt_count"] == 0
+ assert summary["estimated"]["baseline_tokens"] == 150
+ assert summary["estimated"]["emitted_tokens"] == 60
+ assert summary["estimated"]["saved_tokens"] == 90
+ assert {row["token_counter"] for row in summary["estimated"]["by_token_counter"]} == {
+ "engraphis.regex.v1", "estimate_tokens"
+ }
+ assert summary["period"] == {"from_ts": 99, "to_ts": 121}
+ all_time = store.context_savings(workspace_id=wid, repo_id=rid)
+ assert all_time["estimated"]["unclassified_receipt_count"] == 1
+
+ current = store.context_savings(
+ workspace_id=wid, repo_id=rid, release_version="1.5"
+ )
+ assert current["receipt_count"] == 2
+ assert current["usage_receipt_count"] == 2
+ assert current["estimated"]["eligible_receipt_count"] == 1
+ assert current["estimated"]["saved_tokens"] == 60
+ assert current["estimated"]["by_basis"][0]["basis"] == "history_retrieval"
+
+ with pytest.raises(ValueError, match="semantic version"):
+ store.context_savings(workspace_id=wid, release_version="not-a-release")
+
+
+def test_service_context_savings_filters_and_new_receipts_are_versioned():
+ service = MemoryService.create(":memory:", graph_extractor="none")
+ service.remember("Versioned context delivery.", workspace="versioned", scope="workspace")
+ service.recall(
+ "context delivery",
+ workspace="versioned",
+ token_budget=32,
+ response_mode="compact",
+ intent="recall_context",
+ )
+ receipt = service.receipt_log(workspace="versioned")["entries"][0]
+ usage = receipt["metadata"]["token_usage"]
+ assert usage["release_version"] == __version__
+ assert usage["savings_basis"] == "packed_context"
+ assert usage["savings_eligible"] is True
+ filtered = service.context_savings(
+ workspace="versioned", release_version=__version__,
+ from_ts=0, to_ts=9_999_999_999,
+ )
+ assert filtered["estimated"]["eligible_receipt_count"] == 1
+ with pytest.raises(ValidationError, match="semantic version"):
+ service.context_savings(workspace="versioned", release_version="legacy")
+
+
+def test_context_savings_ignores_gateway_copies_and_rejects_noncanonical_estimates():
+ store = Store(":memory:")
+ wid = store.get_or_create_workspace("gateway-savings")
+ authoritative = _usage(100, 40, counter="engraphis.regex.v1")
+ store.record_receipt(
+ "adaptive_context", workspace_id=wid,
+ metadata={"token_usage": authoritative},
+ )
+ store.record_receipt(
+ "smart_gateway", workspace_id=wid,
+ metadata={"token_usage": authoritative},
+ )
+ noncanonical = _usage(80, 20, counter="engraphis.regex.v1")
+ noncanonical["estimated_savings_ratio"] = 0.1
+ store.record_receipt(
+ "adaptive_context", workspace_id=wid,
+ metadata={"token_usage": noncanonical},
+ )
+
+ summary = store.context_savings(workspace_id=wid)
+ assert summary["estimated"]["eligible_receipt_count"] == 1
+ assert summary["estimated"]["saved_tokens"] == 60
+ assert summary["estimated"]["invalid_estimate_count"] == 1
diff --git a/tests/test_update_check.py b/tests/test_update_check.py
index e48e3490..933bc8bd 100644
--- a/tests/test_update_check.py
+++ b/tests/test_update_check.py
@@ -1,302 +1,302 @@
-"""Offline tests for the update-reminder module (engraphis.update_check).
-
-Everything here is deterministic and network-free: version math is pure, and the one
-code path that would hit the network (``_fetch``) is either monkeypatched or exercised
-only on inputs it rejects *before* opening a socket.
-"""
-from __future__ import annotations
-
-import json
-import os
-
-import pytest
-
-from engraphis import update_check as u
-
-
-# ── pure version math ─────────────────────────────────────────────────────────
-@pytest.mark.parametrize("text,expected", [
- ("1.2.3", (1, 2, 3)),
- ("v1.2.3", (1, 2, 3)),
- (" V2.0 ", (2, 0)),
- ("1.2.3-rc1", (1, 2, 3)),
- ("1.0.0+build.5", (1, 0, 0)),
- ("10.4", (10, 4)),
- ("nightly", None),
- ("", None),
- (None, None),
- (123, None),
-])
-def test_parse_version(text, expected):
- assert u.parse_version(text) == expected
-
-
-@pytest.mark.parametrize("text", [
- "1." + "9" * 1000,
- ".".join(["1"] * (u._MAX_VERSION_PARTS + 1)),
-])
-def test_parse_version_rejects_pathological_numeric_versions(text):
- assert u.parse_version(text) is None
-
-
-@pytest.mark.parametrize("latest,current,newer", [
- ("1.1.0", "1.0.0", True),
- ("1.0.1", "1.0.0", True),
- ("2.0", "1.9.9", True),
- ("1.0.0", "1.0.0", False), # equal is not newer
- ("1.0", "1.0.0", False), # zero-padded equal
- ("0.9.9", "1.0.0", False),
- ("v1.2.0", "1.1.5", True), # tolerates the v prefix on both sides
- ("garbage", "1.0.0", False), # unparseable → never newer
- ("1.0.0", "garbage", False),
-])
-def test_is_newer(latest, current, newer):
- assert u.is_newer(latest, current) is newer
-
-
-# ── payload normalization ─────────────────────────────────────────────────────
-def test_parse_github_release():
- got = u._parse_release_payload({
- "tag_name": "v1.4.0", "html_url": "https://example/releases/tag/v1.4.0",
- "draft": False, "prerelease": False,
- })
- assert got == {"version": "v1.4.0", "url": "https://example/releases/tag/v1.4.0"}
-
-
-def test_parse_github_rejects_draft_and_prerelease():
- assert u._parse_release_payload({"tag_name": "v2", "draft": True}) is None
- assert u._parse_release_payload({"tag_name": "v2", "prerelease": True}) is None
-
-
-def test_parse_pypi_payload():
- got = u._parse_release_payload({"info": {"version": "1.5.0"}})
- assert got["version"] == "1.5.0"
- assert "1.5.0" in got["url"]
-
-
-def test_parse_generic_and_garbage():
- assert u._parse_release_payload({"version": "3.0", "url": "https://x/y"}) == {
- "version": "3.0", "url": "https://x/y"}
- assert u._parse_release_payload({"nope": 1}) is None
- assert u._parse_release_payload("not a dict") is None
-
-
-# ── network guard (no socket opened for a bad scheme/host) ────────────────────
-@pytest.mark.parametrize("url", [
- "http://example.com/releases", # plain http, non-loopback
- "ftp://example.com/x",
- "file:///etc/passwd",
- "https://user@example.com/releases",
- "https://[::1/releases",
- "https://example.com\\@127.0.0.1/releases",
-])
-def test_fetch_rejects_unsafe_schemes(url):
- assert u._fetch(url, timeout=0.01) is None
-
-
-def test_fetch_rejects_dns_loopback_alias_before_opening(monkeypatch):
- monkeypatch.setattr(
- u, "build_pinned_https_opener",
- lambda *args, **kwargs: pytest.fail("a DNS alias must not reach an HTTP opener"),
- )
- assert u._fetch("http://localhost/latest", timeout=0.01) is None
-
-
-# ── endpoint / explicit opt-in configuration ──────────────────────────────────
-def test_endpoint_default_and_overrides(monkeypatch):
- monkeypatch.delenv("ENGRAPHIS_UPDATE_URL", raising=False)
- monkeypatch.delenv("ENGRAPHIS_UPDATE_REPO", raising=False)
- assert u._endpoint() == "https://api.github.com/repos/%s/releases/latest" % u.DEFAULT_REPO
- monkeypatch.setenv("ENGRAPHIS_UPDATE_REPO", "acme/thing")
- assert u._endpoint().endswith("/repos/acme/thing/releases/latest")
- monkeypatch.setenv("ENGRAPHIS_UPDATE_URL", "https://mirror/latest.json")
- assert u._endpoint() == "https://mirror/latest.json" # explicit URL wins over repo
-
-
-@pytest.mark.parametrize("value", [
- None, "0", "false", "no", "off", "disable", "disabled",
- "treu", "enabled-ish", "2", "random",
-])
-def test_unset_false_like_and_misspelled_values_stay_offline(monkeypatch, value):
- if value is None:
- monkeypatch.delenv("ENGRAPHIS_UPDATE_CHECK", raising=False)
- else:
- monkeypatch.setenv("ENGRAPHIS_UPDATE_CHECK", value)
- assert u.enabled() is False
-
- # Every non-affirmative value must keep check() from opening a socket.
- monkeypatch.setattr(u, "_fetch", lambda *a, **k: pytest.fail("must not hit network"))
- snap = u.check()
- assert snap == u._disabled_snapshot()
- assert u.notice_line(snap) is None
-
-
-@pytest.mark.parametrize("value", ["1", "true", "yes", "on", "enable", "enabled"])
-def test_recognized_explicit_opt_in_values(monkeypatch, value):
- monkeypatch.setenv("ENGRAPHIS_UPDATE_CHECK", value)
- assert u.enabled() is True
-
-
-# ── cache + snapshot behavior ─────────────────────────────────────────────────
-@pytest.fixture
-def cache(tmp_path, monkeypatch):
- """Isolate the on-disk cache and force checks enabled with a known endpoint."""
- path = tmp_path / "update.json"
- monkeypatch.setenv("ENGRAPHIS_UPDATE_CACHE", str(path))
- monkeypatch.setenv("ENGRAPHIS_UPDATE_CHECK", "1")
- monkeypatch.setenv("ENGRAPHIS_UPDATE_URL", "https://example.test/latest")
- return path
-
-
-def test_check_fetches_writes_cache_and_reports_update(cache, monkeypatch):
- monkeypatch.setattr(u, "CURRENT_VERSION", "1.0.0")
- monkeypatch.setattr(u, "_fetch",
- lambda url, timeout: {"version": "1.4.0", "url": "https://rel/1.4.0"})
- snap = u.check(force=True)
- assert snap["update_available"] is True
- assert snap["latest"] == "1.4.0" and snap["current"] == "1.0.0"
- assert snap["url"] == "https://rel/1.4.0"
- # cache persisted
- saved = json.loads(cache.read_text())
- assert saved["latest"] == "1.4.0" and saved["checked_at"] > 0
-
-
-def test_fresh_cache_short_circuits_network(cache, monkeypatch):
- monkeypatch.setattr(u, "CURRENT_VERSION", "1.0.0")
- u._write_cache("1.3.0", "https://rel/1.3.0")
- monkeypatch.setattr(u, "_fetch", lambda *a, **k: pytest.fail("fresh cache must not refetch"))
- snap = u.check() # not forced → should use the fresh cache
- assert snap["latest"] == "1.3.0" and snap["update_available"] is True
-
-
-def test_upgrade_clears_banner_without_ttl_wait(cache, monkeypatch):
- """After the user upgrades, a still-fresh cache whose ``latest`` == installed version
- must report no update — update_available is recomputed against the live version."""
- u._write_cache("1.4.0", "https://rel/1.4.0")
- monkeypatch.setattr(u, "CURRENT_VERSION", "1.4.0") # simulate the just-installed upgrade
- monkeypatch.setattr(u, "_fetch", lambda *a, **k: pytest.fail("no network needed"))
- snap = u.check()
- assert snap["update_available"] is False
-
-
-def test_fetch_failure_preserves_last_good(cache, monkeypatch):
- monkeypatch.setattr(u, "CURRENT_VERSION", "1.0.0")
- u._write_cache("1.4.0", "https://rel/1.4.0")
- # Expire the cache so check() attempts a refresh, then have the network fail.
- stale = json.loads(cache.read_text())
- stale["checked_at"] = 0.0
- cache.write_text(json.dumps(stale))
- monkeypatch.setattr(u, "_fetch", lambda *a, **k: None)
- snap = u.check()
- assert snap["latest"] == "1.4.0" and snap["update_available"] is True # last good kept
-
-
-def test_unexpected_fetch_failure_is_fail_silent(cache, monkeypatch):
- stale = {"latest": "1.4.0", "url": "https://rel/1.4.0", "checked_at": 0.0}
- cache.write_text(json.dumps(stale))
-
- def fail(*_args, **_kwargs):
- raise RuntimeError("provider detail must not escape")
-
- monkeypatch.setattr(u, "_fetch", fail)
- snap = u.check()
-
- assert snap["latest"] == "1.4.0"
- assert snap["error"] == "update check unavailable"
-
-def test_snapshot_is_non_blocking(cache, monkeypatch):
- monkeypatch.setattr(u, "CURRENT_VERSION", "1.0.0")
- called = {"bg": False}
- monkeypatch.setattr(u, "refresh_in_background", lambda *a, **k: called.__setitem__("bg", True))
- monkeypatch.setattr(u, "_fetch", lambda *a, **k: pytest.fail("snapshot must not fetch inline"))
- snap = u.snapshot() # empty cache → returns immediately, schedules a background refresh
- assert snap["update_available"] is False
- assert called["bg"] is True
-
-
-@pytest.mark.parametrize("checked_at", [
- [1], {"value": 1}, "nan", "inf", "-inf",
-])
-def test_malformed_cache_timestamp_is_fail_silent(cache, monkeypatch, checked_at):
- cache.write_text(json.dumps({"latest": "2.0.0", "checked_at": checked_at}))
- monkeypatch.setattr(u, "refresh_in_background", lambda *args, **kwargs: None)
-
- snap = u.snapshot()
-
- assert snap["checked_at"] == 0.0
-
-
-def test_oversized_cache_is_ignored(cache):
- cache.write_text("x" * (u._MAX_CACHE_BYTES + 1))
- assert u._read_cache() == {}
-
-
-def test_linked_cache_is_ignored_and_never_overwrites_target(cache):
- victim = cache.with_name("victim.json")
- victim.write_text("do not replace")
- try:
- cache.symlink_to(victim)
- except (NotImplementedError, OSError):
- try:
- os.link(victim, cache)
- except OSError:
- pytest.skip("this platform cannot create a link for the cache test")
-
- assert u._read_cache() == {}
- u._write_cache("9.9.9", "https://example.test/release")
- assert victim.read_text() == "do not replace"
-
-
-def test_notice_line(monkeypatch):
- line = u.notice_line({"enabled": True, "update_available": True,
- "latest": "1.4.0", "current": "1.0.0", "url": "https://rel/1.4.0"})
- assert "1.4.0" in line and "1.0.0" in line and "pip install -U engraphis" in line
- assert u.notice_line({"enabled": True, "update_available": False}) is None
-
-
-def test_cli_notice_uses_the_non_blocking_snapshot_and_is_fail_silent(monkeypatch):
- seen = []
- monkeypatch.setenv("ENGRAPHIS_UPDATE_CHECK", "1")
- monkeypatch.setattr(u, "snapshot", lambda: {
- "enabled": True, "update_available": True, "latest": "1.4.0", "current": "1.0.0",
- "url": "https://rel/1.4.0",
- })
- monkeypatch.setattr(u, "check", lambda **_kwargs: pytest.fail("CLI must not check inline"))
- u.emit_cli_notice(seen.append)
- assert seen and "1.4.0" in seen[0]
-
- monkeypatch.setattr(u, "snapshot", lambda: (_ for _ in ()).throw(RuntimeError("offline")))
- u.emit_cli_notice(seen.append)
- assert len(seen) == 1
-
-
-def test_primary_ledger_renders_the_update_snapshot():
- root = __import__("pathlib").Path(__file__).resolve().parents[1] / "engraphis" / "dashboard_assets"
- html = (root / "index.html").read_text(encoding="utf-8")
- script = (root / "ledger.js").read_text(encoding="utf-8")
- css = (root / "ledger.css").read_text(encoding="utf-8")
- assert 'id="update-banner"' in html
- assert "renderUpdateBanner(bootstrap.update)" in script
- assert "pip install -U engraphis" in script
- assert ".update-banner" in css
-
-
-def test_api_update_endpoint(monkeypatch):
- pytest.importorskip("fastapi", reason="v2_api requires fastapi (extras)")
- from engraphis.routes import v2_api
- monkeypatch.setattr(u, "snapshot",
- lambda: {"enabled": True, "update_available": True, "latest": "1.4.0"})
- out = v2_api.api_update(force=False)
- assert out["update_available"] is True and out["latest"] == "1.4.0"
-
-
-def test_api_update_never_raises(monkeypatch):
- pytest.importorskip("fastapi", reason="v2_api requires fastapi (extras)")
- from engraphis.routes import v2_api
-
- def boom():
- raise RuntimeError("nope")
-
- monkeypatch.setattr(u, "snapshot", boom)
- out = v2_api.api_update(force=False)
- assert out == {"enabled": False, "update_available": False}
+"""Offline tests for the update-reminder module (engraphis.update_check).
+
+Everything here is deterministic and network-free: version math is pure, and the one
+code path that would hit the network (``_fetch``) is either monkeypatched or exercised
+only on inputs it rejects *before* opening a socket.
+"""
+from __future__ import annotations
+
+import json
+import os
+
+import pytest
+
+from engraphis import update_check as u
+
+
+# ── pure version math ─────────────────────────────────────────────────────────
+@pytest.mark.parametrize("text,expected", [
+ ("1.2.3", (1, 2, 3)),
+ ("v1.2.3", (1, 2, 3)),
+ (" V2.0 ", (2, 0)),
+ ("1.2.3-rc1", (1, 2, 3)),
+ ("1.0.0+build.5", (1, 0, 0)),
+ ("10.4", (10, 4)),
+ ("nightly", None),
+ ("", None),
+ (None, None),
+ (123, None),
+])
+def test_parse_version(text, expected):
+ assert u.parse_version(text) == expected
+
+
+@pytest.mark.parametrize("text", [
+ "1." + "9" * 1000,
+ ".".join(["1"] * (u._MAX_VERSION_PARTS + 1)),
+])
+def test_parse_version_rejects_pathological_numeric_versions(text):
+ assert u.parse_version(text) is None
+
+
+@pytest.mark.parametrize("latest,current,newer", [
+ ("1.1.0", "1.0.0", True),
+ ("1.0.1", "1.0.0", True),
+ ("2.0", "1.9.9", True),
+ ("1.0.0", "1.0.0", False), # equal is not newer
+ ("1.0", "1.0.0", False), # zero-padded equal
+ ("0.9.9", "1.0.0", False),
+ ("v1.2.0", "1.1.5", True), # tolerates the v prefix on both sides
+ ("garbage", "1.0.0", False), # unparseable → never newer
+ ("1.0.0", "garbage", False),
+])
+def test_is_newer(latest, current, newer):
+ assert u.is_newer(latest, current) is newer
+
+
+# ── payload normalization ─────────────────────────────────────────────────────
+def test_parse_github_release():
+ got = u._parse_release_payload({
+ "tag_name": "v1.4.0", "html_url": "https://example/releases/tag/v1.4.0",
+ "draft": False, "prerelease": False,
+ })
+ assert got == {"version": "v1.4.0", "url": "https://example/releases/tag/v1.4.0"}
+
+
+def test_parse_github_rejects_draft_and_prerelease():
+ assert u._parse_release_payload({"tag_name": "v2", "draft": True}) is None
+ assert u._parse_release_payload({"tag_name": "v2", "prerelease": True}) is None
+
+
+def test_parse_pypi_payload():
+ got = u._parse_release_payload({"info": {"version": "1.5"}})
+ assert got["version"] == "1.5"
+ assert "1.5" in got["url"]
+
+
+def test_parse_generic_and_garbage():
+ assert u._parse_release_payload({"version": "3.0", "url": "https://x/y"}) == {
+ "version": "3.0", "url": "https://x/y"}
+ assert u._parse_release_payload({"nope": 1}) is None
+ assert u._parse_release_payload("not a dict") is None
+
+
+# ── network guard (no socket opened for a bad scheme/host) ────────────────────
+@pytest.mark.parametrize("url", [
+ "http://example.com/releases", # plain http, non-loopback
+ "ftp://example.com/x",
+ "file:///etc/passwd",
+ "https://user@example.com/releases",
+ "https://[::1/releases",
+ "https://example.com\\@127.0.0.1/releases",
+])
+def test_fetch_rejects_unsafe_schemes(url):
+ assert u._fetch(url, timeout=0.01) is None
+
+
+def test_fetch_rejects_dns_loopback_alias_before_opening(monkeypatch):
+ monkeypatch.setattr(
+ u, "build_pinned_https_opener",
+ lambda *args, **kwargs: pytest.fail("a DNS alias must not reach an HTTP opener"),
+ )
+ assert u._fetch("http://localhost/latest", timeout=0.01) is None
+
+
+# ── endpoint / explicit opt-in configuration ──────────────────────────────────
+def test_endpoint_default_and_overrides(monkeypatch):
+ monkeypatch.delenv("ENGRAPHIS_UPDATE_URL", raising=False)
+ monkeypatch.delenv("ENGRAPHIS_UPDATE_REPO", raising=False)
+ assert u._endpoint() == "https://api.github.com/repos/%s/releases/latest" % u.DEFAULT_REPO
+ monkeypatch.setenv("ENGRAPHIS_UPDATE_REPO", "acme/thing")
+ assert u._endpoint().endswith("/repos/acme/thing/releases/latest")
+ monkeypatch.setenv("ENGRAPHIS_UPDATE_URL", "https://mirror/latest.json")
+ assert u._endpoint() == "https://mirror/latest.json" # explicit URL wins over repo
+
+
+@pytest.mark.parametrize("value", [
+ None, "0", "false", "no", "off", "disable", "disabled",
+ "treu", "enabled-ish", "2", "random",
+])
+def test_unset_false_like_and_misspelled_values_stay_offline(monkeypatch, value):
+ if value is None:
+ monkeypatch.delenv("ENGRAPHIS_UPDATE_CHECK", raising=False)
+ else:
+ monkeypatch.setenv("ENGRAPHIS_UPDATE_CHECK", value)
+ assert u.enabled() is False
+
+ # Every non-affirmative value must keep check() from opening a socket.
+ monkeypatch.setattr(u, "_fetch", lambda *a, **k: pytest.fail("must not hit network"))
+ snap = u.check()
+ assert snap == u._disabled_snapshot()
+ assert u.notice_line(snap) is None
+
+
+@pytest.mark.parametrize("value", ["1", "true", "yes", "on", "enable", "enabled"])
+def test_recognized_explicit_opt_in_values(monkeypatch, value):
+ monkeypatch.setenv("ENGRAPHIS_UPDATE_CHECK", value)
+ assert u.enabled() is True
+
+
+# ── cache + snapshot behavior ─────────────────────────────────────────────────
+@pytest.fixture
+def cache(tmp_path, monkeypatch):
+ """Isolate the on-disk cache and force checks enabled with a known endpoint."""
+ path = tmp_path / "update.json"
+ monkeypatch.setenv("ENGRAPHIS_UPDATE_CACHE", str(path))
+ monkeypatch.setenv("ENGRAPHIS_UPDATE_CHECK", "1")
+ monkeypatch.setenv("ENGRAPHIS_UPDATE_URL", "https://example.test/latest")
+ return path
+
+
+def test_check_fetches_writes_cache_and_reports_update(cache, monkeypatch):
+ monkeypatch.setattr(u, "CURRENT_VERSION", "1.0.0")
+ monkeypatch.setattr(u, "_fetch",
+ lambda url, timeout: {"version": "1.4.0", "url": "https://rel/1.4.0"})
+ snap = u.check(force=True)
+ assert snap["update_available"] is True
+ assert snap["latest"] == "1.4.0" and snap["current"] == "1.0.0"
+ assert snap["url"] == "https://rel/1.4.0"
+ # cache persisted
+ saved = json.loads(cache.read_text())
+ assert saved["latest"] == "1.4.0" and saved["checked_at"] > 0
+
+
+def test_fresh_cache_short_circuits_network(cache, monkeypatch):
+ monkeypatch.setattr(u, "CURRENT_VERSION", "1.0.0")
+ u._write_cache("1.3.0", "https://rel/1.3.0")
+ monkeypatch.setattr(u, "_fetch", lambda *a, **k: pytest.fail("fresh cache must not refetch"))
+ snap = u.check() # not forced → should use the fresh cache
+ assert snap["latest"] == "1.3.0" and snap["update_available"] is True
+
+
+def test_upgrade_clears_banner_without_ttl_wait(cache, monkeypatch):
+ """After the user upgrades, a still-fresh cache whose ``latest`` == installed version
+ must report no update — update_available is recomputed against the live version."""
+ u._write_cache("1.4.0", "https://rel/1.4.0")
+ monkeypatch.setattr(u, "CURRENT_VERSION", "1.4.0") # simulate the just-installed upgrade
+ monkeypatch.setattr(u, "_fetch", lambda *a, **k: pytest.fail("no network needed"))
+ snap = u.check()
+ assert snap["update_available"] is False
+
+
+def test_fetch_failure_preserves_last_good(cache, monkeypatch):
+ monkeypatch.setattr(u, "CURRENT_VERSION", "1.0.0")
+ u._write_cache("1.4.0", "https://rel/1.4.0")
+ # Expire the cache so check() attempts a refresh, then have the network fail.
+ stale = json.loads(cache.read_text())
+ stale["checked_at"] = 0.0
+ cache.write_text(json.dumps(stale))
+ monkeypatch.setattr(u, "_fetch", lambda *a, **k: None)
+ snap = u.check()
+ assert snap["latest"] == "1.4.0" and snap["update_available"] is True # last good kept
+
+
+def test_unexpected_fetch_failure_is_fail_silent(cache, monkeypatch):
+ stale = {"latest": "1.4.0", "url": "https://rel/1.4.0", "checked_at": 0.0}
+ cache.write_text(json.dumps(stale))
+
+ def fail(*_args, **_kwargs):
+ raise RuntimeError("provider detail must not escape")
+
+ monkeypatch.setattr(u, "_fetch", fail)
+ snap = u.check()
+
+ assert snap["latest"] == "1.4.0"
+ assert snap["error"] == "update check unavailable"
+
+def test_snapshot_is_non_blocking(cache, monkeypatch):
+ monkeypatch.setattr(u, "CURRENT_VERSION", "1.0.0")
+ called = {"bg": False}
+ monkeypatch.setattr(u, "refresh_in_background", lambda *a, **k: called.__setitem__("bg", True))
+ monkeypatch.setattr(u, "_fetch", lambda *a, **k: pytest.fail("snapshot must not fetch inline"))
+ snap = u.snapshot() # empty cache → returns immediately, schedules a background refresh
+ assert snap["update_available"] is False
+ assert called["bg"] is True
+
+
+@pytest.mark.parametrize("checked_at", [
+ [1], {"value": 1}, "nan", "inf", "-inf",
+])
+def test_malformed_cache_timestamp_is_fail_silent(cache, monkeypatch, checked_at):
+ cache.write_text(json.dumps({"latest": "2.0.0", "checked_at": checked_at}))
+ monkeypatch.setattr(u, "refresh_in_background", lambda *args, **kwargs: None)
+
+ snap = u.snapshot()
+
+ assert snap["checked_at"] == 0.0
+
+
+def test_oversized_cache_is_ignored(cache):
+ cache.write_text("x" * (u._MAX_CACHE_BYTES + 1))
+ assert u._read_cache() == {}
+
+
+def test_linked_cache_is_ignored_and_never_overwrites_target(cache):
+ victim = cache.with_name("victim.json")
+ victim.write_text("do not replace")
+ try:
+ cache.symlink_to(victim)
+ except (NotImplementedError, OSError):
+ try:
+ os.link(victim, cache)
+ except OSError:
+ pytest.skip("this platform cannot create a link for the cache test")
+
+ assert u._read_cache() == {}
+ u._write_cache("9.9.9", "https://example.test/release")
+ assert victim.read_text() == "do not replace"
+
+
+def test_notice_line(monkeypatch):
+ line = u.notice_line({"enabled": True, "update_available": True,
+ "latest": "1.4.0", "current": "1.0.0", "url": "https://rel/1.4.0"})
+ assert "1.4.0" in line and "1.0.0" in line and "pip install -U engraphis" in line
+ assert u.notice_line({"enabled": True, "update_available": False}) is None
+
+
+def test_cli_notice_uses_the_non_blocking_snapshot_and_is_fail_silent(monkeypatch):
+ seen = []
+ monkeypatch.setenv("ENGRAPHIS_UPDATE_CHECK", "1")
+ monkeypatch.setattr(u, "snapshot", lambda: {
+ "enabled": True, "update_available": True, "latest": "1.4.0", "current": "1.0.0",
+ "url": "https://rel/1.4.0",
+ })
+ monkeypatch.setattr(u, "check", lambda **_kwargs: pytest.fail("CLI must not check inline"))
+ u.emit_cli_notice(seen.append)
+ assert seen and "1.4.0" in seen[0]
+
+ monkeypatch.setattr(u, "snapshot", lambda: (_ for _ in ()).throw(RuntimeError("offline")))
+ u.emit_cli_notice(seen.append)
+ assert len(seen) == 1
+
+
+def test_primary_ledger_renders_the_update_snapshot():
+ root = __import__("pathlib").Path(__file__).resolve().parents[1] / "engraphis" / "dashboard_assets"
+ html = (root / "index.html").read_text(encoding="utf-8")
+ script = (root / "ledger.js").read_text(encoding="utf-8")
+ css = (root / "ledger.css").read_text(encoding="utf-8")
+ assert 'id="update-banner"' in html
+ assert "renderUpdateBanner(bootstrap.update)" in script
+ assert "pip install -U engraphis" in script
+ assert ".update-banner" in css
+
+
+def test_api_update_endpoint(monkeypatch):
+ pytest.importorskip("fastapi", reason="v2_api requires fastapi (extras)")
+ from engraphis.routes import v2_api
+ monkeypatch.setattr(u, "snapshot",
+ lambda: {"enabled": True, "update_available": True, "latest": "1.4.0"})
+ out = v2_api.api_update(force=False)
+ assert out["update_available"] is True and out["latest"] == "1.4.0"
+
+
+def test_api_update_never_raises(monkeypatch):
+ pytest.importorskip("fastapi", reason="v2_api requires fastapi (extras)")
+ from engraphis.routes import v2_api
+
+ def boom():
+ raise RuntimeError("nope")
+
+ monkeypatch.setattr(u, "snapshot", boom)
+ out = v2_api.api_update(force=False)
+ assert out == {"enabled": False, "update_available": False}