From 2ac520acbec57d1c75277187dae7e6747d2e5d2b Mon Sep 17 00:00:00 2001 From: Robert Date: Sat, 18 Jul 2026 06:07:23 -0300 Subject: [PATCH] Fail CSVAnswers persistence closed --- a2a.yaml | 2 +- agent.py | 79 +++++++++++++++++++++---------- tests/test_full_stack_contract.py | 32 +++++++++++-- 3 files changed, 83 insertions(+), 30 deletions(-) diff --git a/a2a.yaml b/a2a.yaml index a4ffff1..be29f1e 100644 --- a/a2a.yaml +++ b/a2a.yaml @@ -1,5 +1,5 @@ name: csv-answers-studio-v1 -version: 0.1.0 +version: 0.1.1 entrypoint: agent:CsvAnswersStudioV1 expose: public: false diff --git a/agent.py b/agent.py index cba89c0..94720f9 100644 --- a/agent.py +++ b/agent.py @@ -12,7 +12,6 @@ import io import json import os import re -from datetime import datetime, timezone from decimal import Decimal, InvalidOperation from typing import Annotated, Any @@ -40,6 +39,13 @@ MAX_CELL_CHARS = 2_000 CSV_MEDIA_TYPES = {"text/csv", "application/csv", "text/plain", "application/vnd.ms-excel"} SAFE_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:-]{0,127}$") +AnalysisId = Annotated[ + str, + Field(min_length=1, max_length=128, pattern=r"^[A-Za-z0-9][A-Za-z0-9_.:-]{0,127}$"), +] +CsvText = Annotated[str, Field(min_length=1, max_length=MAX_CSV_BYTES)] +Question = Annotated[str, Field(min_length=1, max_length=500)] + # Local-only fallback lets sandbox/unit tests prove behavior without receiving # managed DATABASE_URL. Hosted deployments use managed Postgres exclusively. _LOCAL_ANALYSES: dict[tuple[str, str], dict[str, Any]] = {} @@ -88,7 +94,7 @@ class CsvAnswersStudioV1(A2AAgent[CsvAnswersStudioV1Config, PlatformUserAuth]): "highest numeric value, persist the grounded answer in user-scoped " "managed Postgres, and reopen it from a polished one-page app." ) - version = "0.1.0" + version = "0.1.1" config_model = CsvAnswersStudioV1Config auth_model = PlatformUserAuth @@ -115,45 +121,52 @@ class CsvAnswersStudioV1(A2AAgent[CsvAnswersStudioV1Config, PlatformUserAuth]): @a2a.tool( description="Analyze bounded pasted CSV text, answer the highest numeric row question, save the result, and persist a receipt.", timeout_seconds=30, - idempotent=True, + idempotent=False, cost_class="cheap", ) async def analyze_csv( self, ctx: RunContext[PlatformUserAuth], - analysis_id: str, - csv_text: str, - question: str, + analysis_id: AnalysisId, + csv_text: CsvText, + question: Question, ) -> CsvAnalysisResult: tenant = _tenant_key(ctx) result = _analyze_csv_text(analysis_id=analysis_id, csv_text=csv_text, question=question) if not result.ok: return result - saved, receipt_id = _persist_analysis_and_receipt( - tenant_key=tenant, - analysis_id=analysis_id, - question=question, - csv_text=csv_text, - result=result.model_dump(mode="json"), - source="pasted_csv", - ) + try: + saved, receipt_id = _persist_analysis_and_receipt( + tenant_key=tenant, + analysis_id=analysis_id, + question=question, + csv_text=csv_text, + result=result.model_dump(mode="json"), + source="pasted_csv", + ) + except Exception: # noqa: BLE001 + return _error( + analysis_id, + "database_error", + "CSVAnswers could not persist this analysis. Try again shortly.", + ) result.saved = saved result.receipt_id = receipt_id - await ctx.emit_progress(f"saved analysis {analysis_id} for {tenant}") + await ctx.emit_progress(f"saved analysis {analysis_id}") return result @a2a.tool( description="Analyze a bounded browser CSV upload encoded as base64 JSON, then save the result and receipt.", timeout_seconds=30, - idempotent=True, + idempotent=False, cost_class="cheap", ) async def analyze_csv_upload_base64( self, ctx: RunContext[PlatformUserAuth], - analysis_id: str, + analysis_id: AnalysisId, upload: BrowserCsvUpload, - question: str, + question: Question, ) -> CsvAnalysisResult: decoded = _decode_browser_upload(upload) if isinstance(decoded, CsvAnalysisResult): @@ -163,13 +176,13 @@ class CsvAnswersStudioV1(A2AAgent[CsvAnswersStudioV1Config, PlatformUserAuth]): @a2a.tool( description="Analyze a typed external FileUpload CSV, then save the result and receipt.", timeout_seconds=30, - idempotent=True, + idempotent=False, cost_class="cheap", ) async def analyze_csv_file( self, ctx: RunContext[PlatformUserAuth], - analysis_id: str, + analysis_id: AnalysisId, file: Annotated[ UploadedFile, FileUpload( @@ -178,7 +191,7 @@ class CsvAnswersStudioV1(A2AAgent[CsvAnswersStudioV1Config, PlatformUserAuth]): description="Bounded CSV file for CSVAnswers analysis.", ), ], - question: str, + question: Question, ) -> CsvAnalysisResult: if file.size_bytes and file.size_bytes > MAX_CSV_BYTES: return _error(analysis_id, "csv_too_large", f"CSV must be at most {MAX_CSV_BYTES} bytes.") @@ -203,10 +216,17 @@ class CsvAnswersStudioV1(A2AAgent[CsvAnswersStudioV1Config, PlatformUserAuth]): async def get_analysis( self, ctx: RunContext[PlatformUserAuth], - analysis_id: str, + analysis_id: AnalysisId, ) -> CsvAnalysisResult: tenant = _tenant_key(ctx) - record = _load_analysis(tenant, analysis_id) + try: + record = _load_analysis(tenant, analysis_id) + except Exception: # noqa: BLE001 + return _error( + analysis_id, + "database_error", + "CSVAnswers could not reload this analysis. Try again shortly.", + ) if record is None: return _error(analysis_id, "not_found", "No saved analysis with that id exists for this signed-in user.") return CsvAnalysisResult.model_validate(record) @@ -224,7 +244,14 @@ class CsvAnswersStudioV1(A2AAgent[CsvAnswersStudioV1Config, PlatformUserAuth]): ) -> CsvAnalysisListResult: tenant = _tenant_key(ctx) safe_limit = min(max(int(limit), 1), 50) - return CsvAnalysisListResult(ok=True, analyses=_list_analyses(tenant, safe_limit)) + try: + return CsvAnalysisListResult(ok=True, analyses=_list_analyses(tenant, safe_limit)) + except Exception: # noqa: BLE001 + return CsvAnalysisListResult( + ok=False, + code="database_error", + message="CSVAnswers could not list saved analyses. Try again shortly.", + ) def _tenant_key(ctx: RunContext[PlatformUserAuth]) -> str: @@ -439,11 +466,12 @@ def _format_decimal(value: Decimal) -> str: def _connect(): url = os.environ.get("DATABASE_URL") if not url: - return None + raise RuntimeError("managed database is not configured") import psycopg return psycopg.connect( url, + connect_timeout=3, options="-c statement_timeout=10000 -c lock_timeout=5000 -c idle_in_transaction_session_timeout=30000", ) @@ -467,7 +495,6 @@ def _persist_analysis_and_receipt( "input_sha256": hashlib.sha256(csv_text.encode("utf-8")).hexdigest(), "question_sha256": hashlib.sha256(question.encode("utf-8")).hexdigest(), "source": source, - "created_at": datetime.now(timezone.utc).isoformat(), "ok": True, } conn = _connect() diff --git a/tests/test_full_stack_contract.py b/tests/test_full_stack_contract.py index 99a0b1e..180a4f2 100644 --- a/tests/test_full_stack_contract.py +++ b/tests/test_full_stack_contract.py @@ -50,6 +50,9 @@ def test_full_stack_product_contract(): assert {"analyze_csv", "get_analysis", "analyze_csv_upload_base64", "analyze_csv_file", "list_analyses"} <= skill_names analyze_schema = next(skill.input_schema for skill in card.skills if skill.name == "analyze_csv") assert set(analyze_schema["required"]) == {"analysis_id", "csv_text", "question"} + assert analyze_schema["properties"]["analysis_id"]["maxLength"] == 128 + assert analyze_schema["properties"]["csv_text"]["maxLength"] == agent_module.MAX_CSV_BYTES + assert analyze_schema["properties"]["question"]["maxLength"] == 500 upload_schema = next(skill.input_schema for skill in card.skills if skill.name == "analyze_csv_file") assert upload_schema["properties"]["file"]["x-a2a-file-upload"]["max_bytes"] == agent_module.MAX_CSV_BYTES databases = card.runtime.platform_resources.databases @@ -60,7 +63,7 @@ def test_full_stack_product_contract(): def test_success_reload_and_receipt_persist_locally(monkeypatch): - monkeypatch.delenv("DATABASE_URL", raising=False) + monkeypatch.setattr(agent_module, "_connect", lambda: None) _reset_local_state_for_tests() app = CsvAnswersStudioV1() @@ -92,7 +95,7 @@ def test_malformed_csv_rejected_deterministically(monkeypatch): def test_browser_base64_bridge(monkeypatch): - monkeypatch.delenv("DATABASE_URL", raising=False) + monkeypatch.setattr(agent_module, "_connect", lambda: None) _reset_local_state_for_tests() app = CsvAnswersStudioV1() upload = { @@ -107,7 +110,7 @@ def test_browser_base64_bridge(monkeypatch): def test_mcp_tools_list_and_call(monkeypatch): - monkeypatch.delenv("DATABASE_URL", raising=False) + monkeypatch.setattr(agent_module, "_connect", lambda: None) _reset_local_state_for_tests() app = CsvAnswersStudioV1() server = MCPServer(app, context_builder=lambda skill_name: make_ctx(f"mcp-{skill_name}")) @@ -140,3 +143,26 @@ def test_mcp_tools_list_and_call(monkeypatch): "params": {"name": "get_analysis", "arguments": {"analysis_id": "studio-csv-v1"}}, })) assert reopened["result"]["structuredContent"]["ok"] is True + + +def test_database_unavailable_fails_closed(monkeypatch): + def unavailable(): + raise RuntimeError("database unavailable") + + monkeypatch.setattr(agent_module, "_connect", unavailable) + app = CsvAnswersStudioV1() + + result = run( + app.invoke( + "analyze_csv", + make_ctx(), + analysis_id="studio-csv-db-failure", + csv_text=SUCCESS_CSV, + question="Which month has the highest revenue?", + ) + ) + + assert result.ok is False + assert result.code == "database_error" + assert result.saved is False + assert "RuntimeError" not in (result.message or "")