bobseares-2/invoice-guard-studio-v1
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Harden InvoiceGuard stateful tools

Browse Source
This commit is contained in:
bobseares@gmail.com
2026-07-18 07:28:00 -03:00
parent 08a4d741ca
commit 1480c952ac
3 changed files with 91 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
a2a.yaml
View File

@@ -1,5 +1,5 @@
name: invoice-guard-studio-v1
version: 0.1.0
version: 0.1.1
entrypoint: agent:InvoiceGuardStudioV1
expose:
public: false

93
agent.py
View File

@@ -9,6 +9,7 @@ small typed tools that are also served through MCP.
from __future__ import annotations
import base64
import asyncio
import csv
import hashlib
import io
@@ -69,7 +70,7 @@ class InvoiceGuardStudioV1(A2AAgent[InvoiceGuardStudioV1Config, PlatformUserAuth
"One-page InvoiceGuard app for deterministic duplicate invoice number "
"and stated-total mismatch review with user-scoped Postgres persistence."
)
version = "0.1.0"
version = "0.1.1"
config_model = InvoiceGuardStudioV1Config
auth_model = PlatformUserAuth
@@ -82,7 +83,7 @@ class InvoiceGuardStudioV1(A2AAgent[InvoiceGuardStudioV1Config, PlatformUserAuth
)
workspace_access = WorkspaceAccess.dynamic(
max_files=8,
allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY),
allowed_modes=(WorkspaceMode.READ_ONLY,),
require_reason=False,
max_total_size_bytes=MAX_UPLOAD_BYTES * MAX_UPLOADS,
)
@@ -102,21 +103,29 @@ class InvoiceGuardStudioV1(A2AAgent[InvoiceGuardStudioV1Config, PlatformUserAuth
@a2a.tool(
description="Review pasted invoice objects, detect duplicate invoice numbers and stated-total mismatches, persist a user-scoped case, and record a production receipt.",
timeout_seconds=30,
idempotent=True,
idempotent=False,
cost_class="deterministic",
)
async def review_invoices(
self,
ctx: RunContext[PlatformUserAuth],
case_id: str,
invoices: list[dict[str, Any]],
case_id: Annotated[
str,
Field(min_length=1, max_length=120, pattern=r"^[A-Za-z0-9_.:-]+$"),
],
invoices: Annotated[list[dict[str, Any]], Field(min_length=1, max_length=MAX_INVOICES)],
) -> dict[str, Any]:
tenant = _tenant_key(ctx)
result = _review_payload(case_id, invoices)
if not result["ok"]:
await ctx.emit_error(result["message"], code=result["code"])
return result
persisted = _persist_case_with_receipt(tenant, result, source="json")
persisted = await asyncio.to_thread(
_persist_case_with_receipt,
tenant,
result,
"json",
)
await ctx.emit_progress(
f"reviewed {result['invoice_count']} invoice(s): {result['duplicate_count']} duplicate group(s), {result['total_mismatch_count']} total mismatch(es)"
)
@@ -125,14 +134,17 @@ class InvoiceGuardStudioV1(A2AAgent[InvoiceGuardStudioV1Config, PlatformUserAuth
@a2a.tool(
description="Review invoices uploaded by browser JSON/base64 bridge, then persist the same review case and receipt as pasted invoices.",
timeout_seconds=30,
idempotent=True,
idempotent=False,
cost_class="deterministic",
)
async def review_invoice_uploads(
self,
ctx: RunContext[PlatformUserAuth],
case_id: str,
documents: list[BrowserDocument],
case_id: Annotated[
str,
Field(min_length=1, max_length=120, pattern=r"^[A-Za-z0-9_.:-]+$"),
],
documents: Annotated[list[BrowserDocument], Field(min_length=1, max_length=MAX_UPLOADS)],
) -> dict[str, Any]:
tenant = _tenant_key(ctx)
parsed = _parse_browser_documents(documents)
@@ -143,19 +155,27 @@ class InvoiceGuardStudioV1(A2AAgent[InvoiceGuardStudioV1Config, PlatformUserAuth
if not result["ok"]:
await ctx.emit_error(result["message"], code=result["code"])
return result
persisted = _persist_case_with_receipt(tenant, result, source="browser_upload")
persisted = await asyncio.to_thread(
_persist_case_with_receipt,
tenant,
result,
"browser_upload",
)
return {**result, "receipt": persisted["receipt"], "saved_at": persisted["saved_at"]}
@a2a.tool(
description="Review a typed Agent API file upload containing JSON, CSV, or plain-text invoice rows.",
timeout_seconds=30,
idempotent=True,
idempotent=False,
cost_class="deterministic",
)
async def review_invoice_file(
self,
ctx: RunContext[PlatformUserAuth],
case_id: str,
case_id: Annotated[
str,
Field(min_length=1, max_length=120, pattern=r"^[A-Za-z0-9_.:-]+$"),
],
document: Annotated[
UploadedFile,
FileUpload(
@@ -179,7 +199,12 @@ class InvoiceGuardStudioV1(A2AAgent[InvoiceGuardStudioV1Config, PlatformUserAuth
if not result["ok"]:
await ctx.emit_error(result["message"], code=result["code"])
return result
persisted = _persist_case_with_receipt(tenant, result, source="file_upload")
persisted = await asyncio.to_thread(
_persist_case_with_receipt,
tenant,
result,
"file_upload",
)
return {**result, "receipt": persisted["receipt"], "saved_at": persisted["saved_at"]}
@a2a.tool(
@@ -191,10 +216,17 @@ class InvoiceGuardStudioV1(A2AAgent[InvoiceGuardStudioV1Config, PlatformUserAuth
async def get_invoice_case(
self,
ctx: RunContext[PlatformUserAuth],
case_id: str,
case_id: Annotated[
str,
Field(min_length=1, max_length=120, pattern=r"^[A-Za-z0-9_.:-]+$"),
],
) -> dict[str, Any]:
tenant = _tenant_key(ctx)
record = _load_case(tenant, case_id)
try:
clean_case_id = _clean_case_id(case_id)
except ValueError as exc:
return _validation_error("invalid_case_id", str(exc), case_id=str(case_id or ""))
record = await asyncio.to_thread(_load_case, tenant, clean_case_id)
if record is None:
return _validation_error("case_not_found", "No invoice review case exists for this signed-in user and case_id.", case_id=case_id)
return record
@@ -202,17 +234,29 @@ class InvoiceGuardStudioV1(A2AAgent[InvoiceGuardStudioV1Config, PlatformUserAuth
@a2a.tool(
description="Record or update a user decision for a saved invoice review case.",
timeout_seconds=15,
idempotent=True,
idempotent=False,
cost_class="deterministic",
)
async def record_decision(
self,
ctx: RunContext[PlatformUserAuth],
case_id: str,
case_id: Annotated[
str,
Field(min_length=1, max_length=120, pattern=r"^[A-Za-z0-9_.:-]+$"),
],
decision: DecisionInput,
) -> dict[str, Any]:
tenant = _tenant_key(ctx)
updated = _persist_decision(tenant, case_id, decision.model_dump())
try:
clean_case_id = _clean_case_id(case_id)
except ValueError as exc:
return _validation_error("invalid_case_id", str(exc), case_id=str(case_id or ""))
updated = await asyncio.to_thread(
_persist_decision,
tenant,
clean_case_id,
decision.model_dump(),
)
if updated is None:
return _validation_error("case_not_found", "Save or review the invoice case before recording decisions.", case_id=case_id)
return updated
@@ -418,7 +462,7 @@ def _receipt_for(tenant: str, case: dict[str, Any], source: str, saved_at: str)
"receipt_id": f"invoice_guard:{case['case_id']}:{digest[:16]}",
"kind": "production_execution_receipt",
"agent": "invoice-guard-studio-v1",
"agent_version": "0.1.0",
"agent_version": "0.1.1",
"skill": "review_invoices",
"tenant_hash": hashlib.sha256(tenant.encode("utf-8")).hexdigest()[:16],
"input_hash": digest,
@@ -533,7 +577,16 @@ def _persist_decision(tenant: str, case_id: str, decision: dict[str, Any]) -> di
""",
(json.dumps(decisions), tenant, clean_case_id),
)
return _load_case(tenant, clean_case_id)
committed = conn.execute(
"""
SELECT case_id, invoice_count, duplicate_count, total_mismatch_count,
invoices, duplicates, total_mismatches, decisions, receipt, created_at, updated_at
FROM invoice_cases
WHERE tenant_key = %s AND case_id = %s
""",
(tenant, clean_case_id),
).fetchone()
return _row_to_case(committed) if committed else None
def _dict_row_factory():

17
tests/test_full_stack_contract.py
View File

@@ -54,6 +54,9 @@ def test_full_stack_product_contract():
assert card.runtime.platform_resources.databases
assert card.runtime.platform_resources.databases[0].migrations.path == "db/migrations"
assert card.workspace_access.enabled is True
assert [mode.value for mode in card.workspace_access.allowed_modes] == ["read_only"]
assert skills["record_decision"].policy.idempotent is False
assert skills["review_invoices"].input_schema["properties"]["case_id"]["maxLength"] == 120
def test_review_success_fixture_is_deterministic():
@@ -130,3 +133,17 @@ def test_acceptance_calls_with_mocked_persistence_reload_and_decision():
updated = asyncio.run(agent.record_decision(ctx, "studio-invoice-v1", DecisionInput(decision_id="d1", decision="hold", note="Investigate duplicate", invoice_number="INV-100")))
assert updated["ok"] is True
assert updated["decisions"][0]["decision"] == "hold"
invalid_reload = asyncio.run(agent.get_invoice_case(ctx, "bad case id"))
assert invalid_reload["ok"] is False
assert invalid_reload["code"] == "invalid_case_id"
invalid_decision = asyncio.run(
agent.record_decision(
ctx,
"bad case id",
DecisionInput(decision_id="d2", decision="hold"),
)
)
assert invalid_decision["ok"] is False
assert invalid_decision["code"] == "invalid_case_id"