Fail CSVAnswers persistence closed

This commit is contained in:
2026-07-18 06:07:23 -03:00
parent c044d9deab
commit 2ac520acbe
3 changed files with 83 additions and 30 deletions

View File

@@ -1,5 +1,5 @@
name: csv-answers-studio-v1 name: csv-answers-studio-v1
version: 0.1.0 version: 0.1.1
entrypoint: agent:CsvAnswersStudioV1 entrypoint: agent:CsvAnswersStudioV1
expose: expose:
public: false public: false

View File

@@ -12,7 +12,6 @@ import io
import json import json
import os import os
import re import re
from datetime import datetime, timezone
from decimal import Decimal, InvalidOperation from decimal import Decimal, InvalidOperation
from typing import Annotated, Any 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"} 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}$") 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 # Local-only fallback lets sandbox/unit tests prove behavior without receiving
# managed DATABASE_URL. Hosted deployments use managed Postgres exclusively. # managed DATABASE_URL. Hosted deployments use managed Postgres exclusively.
_LOCAL_ANALYSES: dict[tuple[str, str], dict[str, Any]] = {} _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 " "highest numeric value, persist the grounded answer in user-scoped "
"managed Postgres, and reopen it from a polished one-page app." "managed Postgres, and reopen it from a polished one-page app."
) )
version = "0.1.0" version = "0.1.1"
config_model = CsvAnswersStudioV1Config config_model = CsvAnswersStudioV1Config
auth_model = PlatformUserAuth auth_model = PlatformUserAuth
@@ -115,45 +121,52 @@ class CsvAnswersStudioV1(A2AAgent[CsvAnswersStudioV1Config, PlatformUserAuth]):
@a2a.tool( @a2a.tool(
description="Analyze bounded pasted CSV text, answer the highest numeric row question, save the result, and persist a receipt.", description="Analyze bounded pasted CSV text, answer the highest numeric row question, save the result, and persist a receipt.",
timeout_seconds=30, timeout_seconds=30,
idempotent=True, idempotent=False,
cost_class="cheap", cost_class="cheap",
) )
async def analyze_csv( async def analyze_csv(
self, self,
ctx: RunContext[PlatformUserAuth], ctx: RunContext[PlatformUserAuth],
analysis_id: str, analysis_id: AnalysisId,
csv_text: str, csv_text: CsvText,
question: str, question: Question,
) -> CsvAnalysisResult: ) -> CsvAnalysisResult:
tenant = _tenant_key(ctx) tenant = _tenant_key(ctx)
result = _analyze_csv_text(analysis_id=analysis_id, csv_text=csv_text, question=question) result = _analyze_csv_text(analysis_id=analysis_id, csv_text=csv_text, question=question)
if not result.ok: if not result.ok:
return result return result
saved, receipt_id = _persist_analysis_and_receipt( try:
tenant_key=tenant, saved, receipt_id = _persist_analysis_and_receipt(
analysis_id=analysis_id, tenant_key=tenant,
question=question, analysis_id=analysis_id,
csv_text=csv_text, question=question,
result=result.model_dump(mode="json"), csv_text=csv_text,
source="pasted_csv", 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.saved = saved
result.receipt_id = receipt_id 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 return result
@a2a.tool( @a2a.tool(
description="Analyze a bounded browser CSV upload encoded as base64 JSON, then save the result and receipt.", description="Analyze a bounded browser CSV upload encoded as base64 JSON, then save the result and receipt.",
timeout_seconds=30, timeout_seconds=30,
idempotent=True, idempotent=False,
cost_class="cheap", cost_class="cheap",
) )
async def analyze_csv_upload_base64( async def analyze_csv_upload_base64(
self, self,
ctx: RunContext[PlatformUserAuth], ctx: RunContext[PlatformUserAuth],
analysis_id: str, analysis_id: AnalysisId,
upload: BrowserCsvUpload, upload: BrowserCsvUpload,
question: str, question: Question,
) -> CsvAnalysisResult: ) -> CsvAnalysisResult:
decoded = _decode_browser_upload(upload) decoded = _decode_browser_upload(upload)
if isinstance(decoded, CsvAnalysisResult): if isinstance(decoded, CsvAnalysisResult):
@@ -163,13 +176,13 @@ class CsvAnswersStudioV1(A2AAgent[CsvAnswersStudioV1Config, PlatformUserAuth]):
@a2a.tool( @a2a.tool(
description="Analyze a typed external FileUpload CSV, then save the result and receipt.", description="Analyze a typed external FileUpload CSV, then save the result and receipt.",
timeout_seconds=30, timeout_seconds=30,
idempotent=True, idempotent=False,
cost_class="cheap", cost_class="cheap",
) )
async def analyze_csv_file( async def analyze_csv_file(
self, self,
ctx: RunContext[PlatformUserAuth], ctx: RunContext[PlatformUserAuth],
analysis_id: str, analysis_id: AnalysisId,
file: Annotated[ file: Annotated[
UploadedFile, UploadedFile,
FileUpload( FileUpload(
@@ -178,7 +191,7 @@ class CsvAnswersStudioV1(A2AAgent[CsvAnswersStudioV1Config, PlatformUserAuth]):
description="Bounded CSV file for CSVAnswers analysis.", description="Bounded CSV file for CSVAnswers analysis.",
), ),
], ],
question: str, question: Question,
) -> CsvAnalysisResult: ) -> CsvAnalysisResult:
if file.size_bytes and file.size_bytes > MAX_CSV_BYTES: 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.") 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( async def get_analysis(
self, self,
ctx: RunContext[PlatformUserAuth], ctx: RunContext[PlatformUserAuth],
analysis_id: str, analysis_id: AnalysisId,
) -> CsvAnalysisResult: ) -> CsvAnalysisResult:
tenant = _tenant_key(ctx) 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: if record is None:
return _error(analysis_id, "not_found", "No saved analysis with that id exists for this signed-in user.") return _error(analysis_id, "not_found", "No saved analysis with that id exists for this signed-in user.")
return CsvAnalysisResult.model_validate(record) return CsvAnalysisResult.model_validate(record)
@@ -224,7 +244,14 @@ class CsvAnswersStudioV1(A2AAgent[CsvAnswersStudioV1Config, PlatformUserAuth]):
) -> CsvAnalysisListResult: ) -> CsvAnalysisListResult:
tenant = _tenant_key(ctx) tenant = _tenant_key(ctx)
safe_limit = min(max(int(limit), 1), 50) 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: def _tenant_key(ctx: RunContext[PlatformUserAuth]) -> str:
@@ -439,11 +466,12 @@ def _format_decimal(value: Decimal) -> str:
def _connect(): def _connect():
url = os.environ.get("DATABASE_URL") url = os.environ.get("DATABASE_URL")
if not url: if not url:
return None raise RuntimeError("managed database is not configured")
import psycopg import psycopg
return psycopg.connect( return psycopg.connect(
url, url,
connect_timeout=3,
options="-c statement_timeout=10000 -c lock_timeout=5000 -c idle_in_transaction_session_timeout=30000", 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(), "input_sha256": hashlib.sha256(csv_text.encode("utf-8")).hexdigest(),
"question_sha256": hashlib.sha256(question.encode("utf-8")).hexdigest(), "question_sha256": hashlib.sha256(question.encode("utf-8")).hexdigest(),
"source": source, "source": source,
"created_at": datetime.now(timezone.utc).isoformat(),
"ok": True, "ok": True,
} }
conn = _connect() conn = _connect()

View File

@@ -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 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") 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 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") 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 assert upload_schema["properties"]["file"]["x-a2a-file-upload"]["max_bytes"] == agent_module.MAX_CSV_BYTES
databases = card.runtime.platform_resources.databases 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): 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() _reset_local_state_for_tests()
app = CsvAnswersStudioV1() app = CsvAnswersStudioV1()
@@ -92,7 +95,7 @@ def test_malformed_csv_rejected_deterministically(monkeypatch):
def test_browser_base64_bridge(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() _reset_local_state_for_tests()
app = CsvAnswersStudioV1() app = CsvAnswersStudioV1()
upload = { upload = {
@@ -107,7 +110,7 @@ def test_browser_base64_bridge(monkeypatch):
def test_mcp_tools_list_and_call(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() _reset_local_state_for_tests()
app = CsvAnswersStudioV1() app = CsvAnswersStudioV1()
server = MCPServer(app, context_builder=lambda skill_name: make_ctx(f"mcp-{skill_name}")) 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"}}, "params": {"name": "get_analysis", "arguments": {"analysis_id": "studio-csv-v1"}},
})) }))
assert reopened["result"]["structuredContent"]["ok"] is True 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 "")