Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
182 changes: 172 additions & 10 deletions examples/harness/provider_voice_environment/import_backend/agent.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,77 @@
app = FastAPI()
_trace = Path(os.getenv("PROVIDER_TRACE_PATH", "/tmp/provider-trace.jsonl"))

_patients = [
{
"patient_id": "pat_001",
"full_name": "Jane Doe",
"date_of_birth": "1985-04-12",
"phone": "+15551234567",
},
{
"patient_id": "pat_002",
"full_name": "John Smith",
"date_of_birth": "1972-09-28",
"phone": "+15559876543",
},
{
"patient_id": "pat_003",
"full_name": "Alice Taylor",
"date_of_birth": "1990-11-05",
"phone": "+15554567890",
},
]
_prescriptions = [
{
"rx_id": "rx_101",
"patient_id": "pat_001",
"medication_name": "Amoxicillin 500mg",
"status": "active",
"refills_remaining": 2,
"is_controlled": False,
},
{
"rx_id": "rx_102",
"patient_id": "pat_001",
"medication_name": "Lisinopril 10mg",
"status": "no_refills_remaining",
"refills_remaining": 0,
"is_controlled": False,
},
{
"rx_id": "rx_103",
"patient_id": "pat_002",
"medication_name": "Zolpidem 5mg",
"status": "controlled_substance",
"refills_remaining": 1,
"is_controlled": True,
},
{
"rx_id": "rx_104",
"patient_id": "pat_002",
"medication_name": "Atorvastatin 20mg",
"status": "expired",
"refills_remaining": 3,
"is_controlled": False,
},
{
"rx_id": "rx_105",
"patient_id": "pat_003",
"medication_name": "Omeprazole 20mg",
"status": "too_soon",
"refills_remaining": 1,
"is_controlled": False,
},
]
_refills = {
"ref_201": {
"refill_id": "ref_201",
"rx_id": "rx_101",
"status": "ready_for_collection",
},
"ref_202": {"refill_id": "ref_202", "rx_id": "rx_105", "status": "processing"},
}


def _record(kind: str, body: Any) -> None:
_trace.parent.mkdir(parents=True, exist_ok=True)
Expand All @@ -33,6 +104,24 @@ def _record(kind: str, body: Any) -> None:
)


def _arguments(body: Any) -> dict[str, Any]:
if not isinstance(body, dict):
return {}
arguments = body.get("args") or body.get("arguments") or body
if isinstance(arguments, str):
try:
arguments = json.loads(arguments)
except ValueError:
return {}
return arguments if isinstance(arguments, dict) else {}


async def _tool_request(request: Request, name: str) -> dict[str, Any]:
body = await request.json()
_record(f"tool.{name}", body)
return _arguments(body)


@app.get("/health")
def health() -> dict[str, bool]:
return {"ok": True}
Expand All @@ -48,21 +137,94 @@ async def provider_events(request: Request) -> dict[str, bool]:
async def record_preference(request: Request) -> dict[str, Any]:
body = await request.json()
_record("tool.record_preference", body)
preference = body.get("preference") if isinstance(body, dict) else None
if preference is None and isinstance(body, dict):
arguments = body.get("args") or body.get("arguments")
if isinstance(arguments, str):
try:
arguments = json.loads(arguments)
except ValueError:
arguments = None
if isinstance(arguments, dict):
preference = arguments.get("preference")
preference = _arguments(body).get("preference")
return {
"recorded": True,
"preference": preference,
}


@app.post("/provider/tools/find_patient")
async def find_patient(request: Request) -> dict[str, Any]:
arguments = await _tool_request(request, "find_patient")
phone = str(arguments.get("phone") or "").strip()
full_name = str(arguments.get("full_name") or "").strip().casefold()
date_of_birth = str(arguments.get("date_of_birth") or "").strip()
for patient in _patients:
phone_match = phone and patient["phone"] == phone
identity_match = (
full_name
and date_of_birth
and patient["full_name"].casefold() == full_name
and patient["date_of_birth"] == date_of_birth
)
if phone_match or identity_match:
return {"found": True, "patient": patient}
return {"found": False, "patient": None}


@app.post("/provider/tools/list_prescriptions")
async def list_prescriptions(request: Request) -> dict[str, Any]:
arguments = await _tool_request(request, "list_prescriptions")
patient_id = str(arguments.get("patient_id") or "").strip()
return {
"patient_id": patient_id,
"prescriptions": [
prescription
for prescription in _prescriptions
if prescription["patient_id"] == patient_id
],
}


@app.post("/provider/tools/request_refill")
async def request_refill(request: Request) -> dict[str, Any]:
arguments = await _tool_request(request, "request_refill")
patient_id = str(arguments.get("patient_id") or "").strip()
rx_id = str(arguments.get("rx_id") or "").strip()
prescription = next(
(
item
for item in _prescriptions
if item["patient_id"] == patient_id and item["rx_id"] == rx_id
),
None,
)
if prescription is None:
return {"accepted": False, "reason": "prescription_not_found"}
refusal = {
"too_soon": "too_soon",
"no_refills_remaining": "no_refills_remaining",
"expired": "prescription_expired",
"controlled_substance": "controlled_medication_requires_prescriber",
}.get(str(prescription["status"]))
if refusal:
return {"accepted": False, "reason": refusal, "rx_id": rx_id}
refill_id = f"ref_{rx_id.removeprefix('rx_')}"
_refills[refill_id] = {
"refill_id": refill_id,
"rx_id": rx_id,
"status": "processing",
}
return {"accepted": True, **_refills[refill_id]}


@app.post("/provider/tools/check_refill_status")
async def check_refill_status(request: Request) -> dict[str, Any]:
arguments = await _tool_request(request, "check_refill_status")
refill_id = str(arguments.get("refill_id") or "").strip()
refill = _refills.get(refill_id)
return {"found": refill is not None, "refill": refill}


@app.post("/provider/tools/transfer_to_pharmacist")
async def transfer_to_pharmacist(request: Request) -> dict[str, Any]:
arguments = await _tool_request(request, "transfer_to_pharmacist")
return {
"transferred": True,
"reason": str(arguments.get("reason") or "clinical_question"),
}


if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=int(os.getenv("PORT", "8080")))
34 changes: 22 additions & 12 deletions src/fi/alk/harness/bundle_author_v2.py
Original file line number Diff line number Diff line change
Expand Up @@ -407,20 +407,24 @@ def _contract_sql_type(declaration: str) -> str | None:
("UNION[", "OPTIONAL[", "LIST[", "DICT[", "MAPPING[", "TUPLE[", "SET[")
)
if language_type:
# Preserve the outer value shape before considering words that describe
# members. Human-readable contracts commonly say things such as
# ``list of dicts with qty (int), cost_basis (float)``; classifying that
# declaration as a number destroys the authored JSON value.
if tokens & {"DICT", "MAPPING", "OBJECT", "JSON", "ANY"}:
return "jsonb"
if tokens & {"LIST", "ARRAY", "TUPLE", "SET"}:
return "jsonb"
if tokens & {"FLOAT", "NUMBER", "DECIMAL", "DOUBLE"}:
return "double precision"
if tokens & {"INT", "INTEGER"}:
return "bigint"
if tokens & {"BOOL", "BOOLEAN"}:
return "boolean"
if tokens & {"DICT", "MAPPING", "OBJECT", "JSON", "ANY"}:
return "jsonb"
if tokens & {"LIST", "ARRAY", "TUPLE", "SET"}:
return "jsonb"
if tokens & {"STR", "STRING"}:
return "text"
patterns = (
(r"^BOOLEAN\b", "boolean"),
(r"^(?:BOOL|BOOLEAN)\b", "boolean"),
(r"^(?:BIGINT|INTEGER|INT|SMALLINT)\b", "bigint"),
(r"^(?:DOUBLE PRECISION|REAL|FLOAT)\b", "double precision"),
(
Expand All @@ -440,17 +444,17 @@ def _contract_sql_type(declaration: str) -> str | None:
# Application contracts often use language-level unions instead of SQL
# declarations. Interpret the whole declaration as a type set, with the wider
# compatible representation winning independently of token order.
if re.match(r"^(?:DICT|MAPPING|OBJECT|JSON|ANY)\b", normalized):
return "jsonb"
if re.match(r"^(?:LIST|ARRAY|TUPLE|SET)\b", normalized):
# Without a proven homogeneous leaf type, JSONB preserves the value shape.
return "jsonb"
if tokens & {"FLOAT", "NUMBER", "DECIMAL", "DOUBLE"}:
return "double precision"
if tokens & {"INT", "INTEGER"}:
return "bigint"
if tokens & {"BOOL", "BOOLEAN"}:
return "boolean"
if tokens & {"DICT", "MAPPING", "OBJECT", "JSON", "ANY"}:
return "jsonb"
if tokens & {"LIST", "ARRAY", "TUPLE", "SET"}:
# Without a proven homogeneous leaf type, JSONB preserves the value shape.
return "jsonb"
if tokens & {"STR", "STRING"}:
return "text"
return None
Expand Down Expand Up @@ -1665,7 +1669,7 @@ def resolve_environment_plan(
environment = {**declared_runtime_environment, **environment}
environment.setdefault(
"LIVEKIT_AGENT_NAME",
"voice-booking-{{JOB_ID}}-w{{WORLD_INDEX}}",
"voice-booking-{{JOB_ID}}-w{{WORLD_INDEX}}",
)
environment.setdefault(
"HARNESS_TOOL_TRACE",
Expand Down Expand Up @@ -2668,7 +2672,13 @@ def _schema_column_types(paths: list[Path]) -> dict[tuple[str, str], str]:
if len(words) < 2:
continue
name = words[0].strip('"')
if name.upper() in ("PRIMARY", "FOREIGN", "UNIQUE", "CHECK", "CONSTRAINT"):
if name.upper() in (
"PRIMARY",
"FOREIGN",
"UNIQUE",
"CHECK",
"CONSTRAINT",
):
continue
kind = words[1].upper()
rest = " ".join(words[1:]).upper()
Expand Down
23 changes: 22 additions & 1 deletion src/fi/alk/harness/chat_worker.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,13 @@
# Model configuration only: no platform callback capability or source-fetch credential.
_CHAT_ENV = frozenset(
{
# Platform-owned model gateway credentials are simulator configuration,
# not target-agent or callback credentials. The isolated worker needs
# them when the Claude SDK wire protocol is routed to a non-Claude model.
"AGENTCC_API_KEY",
"AGENTCC_BASE_URL",
"ALK_CLAUDE_GATEWAY_API_KEY",
"ALK_CLAUDE_GATEWAY_URL",
"ALK_HARNESS",
"ALK_HARNESS_MODEL",
"ALK_HARNESS_THINKING",
Expand Down Expand Up @@ -85,7 +92,11 @@ async def run(
work_directory=context.work_directory / "chat-workers",
)
if "error" in result:
raise CallAborted(f"chat_worker_failed: {result['error']}")
raise CallAborted(
f"chat_worker_failed: {result['error']}",
marker=result.get("error_marker"),
code=result.get("error_code") or "call_failed",
)
artifact_ids = {}
for artifact in result["artifacts"]:
artifact_ids[artifact["id"]] = await self._adapter.upload_artifact(
Expand Down Expand Up @@ -154,6 +165,16 @@ async def main() -> None:
EnvironmentRuntime.model_validate(payload["runtime"]),
)
result = {"outcome": asdict(outcome), "artifacts": collector.artifacts}
except CallAborted as exc:
# CallAborted messages and structured failure fields are already designed for
# result receipts. Preserve them across the worker boundary; returning only the
# exception class made every chat failure indistinguishable and discarded useful
# retry/attribution codes. Arbitrary exceptions remain type-only below.
result = {
"error": str(exc),
"error_code": exc.code,
"error_marker": exc.marker,
}
except Exception as exc: # noqa: BLE001 - return a secret-safe worker failure
result = {"error": type(exc).__name__}
fd = os.open(sys.argv[1], os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
Expand Down
Loading
Loading