import json import os import shutil import sys import ast import subprocess import tempfile import textwrap import threading import time import unittest from contextlib import redirect_stdout from io import StringIO from unittest.mock import patch sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) import tide from tide import cli as tide_cli from tide import packs as tide_packs from tide import policy as tide_policy from tide import store as tide_store def events(project): return tide_store.read_jsonl(project, tide_store.EVENTS_FILE) class TideRuntimeTest(unittest.TestCase): def setUp(self): self.project = tempfile.mkdtemp(prefix="tide-ledger-test-") tide.configure(project=self.project, agent_id="test-agent", workflow_id="test-flow", environment="prod", on_approval="raise", approval_timeout=5, approval_poll=0.05) def tearDown(self): shutil.rmtree(self.project, ignore_errors=True) def test_policy_matrix_defaults(self): pol = tide_policy.DEFAULT_POLICY self.assertEqual(tide_policy.decide(pol, "t", "read_only", "prod")[0], "allow") self.assertEqual(tide_policy.decide(pol, "t", "internal_write", "prod")[0], "require_approval") self.assertEqual(tide_policy.decide(pol, "t", "internal_write", "dev")[0], "allow_with_log") self.assertEqual(tide_policy.decide(pol, "t", "destructive", "prod")[0], "block") def test_policy_fails_closed(self): pol = tide_policy.DEFAULT_POLICY self.assertEqual(tide_policy.decide(pol, "t", "not_a_class", "prod")[0], "block") self.assertEqual(tide_policy.decide({"decisions": {}}, "t", "read_only", "prod")[0], "block") def test_per_tool_override_wins(self): pol = json.loads(json.dumps(tide_policy.DEFAULT_POLICY)) pol["tools"] = {"crm.nuke": {"decision": "block", "reason": "never"}} decision, reason = tide_policy.decide(pol, "crm.nuke", "read_only", "prod") self.assertEqual(decision, "block") self.assertEqual(reason, "never") def test_dynamic_risk_class(self): def classify(named): sql = named["sql"].lower() return "destructive" if "drop " in sql else "read_only" @tide.controlled_tool(name="sql.query", risk_class=classify) def query(sql): return "ok" self.assertEqual(query("select * from customers"), "ok") with self.assertRaises(tide.LedgerBlocked): query("drop table customers") def test_allow_records_event_with_hashes_and_redaction(self): @tide.controlled_tool(name="crm.get_lead", risk_class="read_only", capture=["lead_id"]) def get_lead(lead_id, secret_note): return {"lead_id": lead_id, "name": "Ada"} result = get_lead("L-1", secret_note="do not store") self.assertEqual(result["name"], "Ada") e = events(self.project)[0] self.assertEqual(e["policy_decision"], "allow") self.assertEqual(e["captured"], {"lead_id": "L-1"}) self.assertNotIn("do not store", json.dumps(e)) self.assertTrue(e["input_hash"].startswith("sha256:")) self.assertTrue(e["output_hash"].startswith("sha256:")) self.assertIsNone(e["error_code"]) def test_block_raises_and_records(self): @tide.controlled_tool(name="crm.delete_account", risk_class="destructive") def delete_account(account_id): raise AssertionError("must never run") with self.assertRaises(tide.LedgerBlocked): delete_account("A-1") e = events(self.project)[0] self.assertEqual(e["policy_decision"], "block") self.assertEqual(e["error_code"], "blocked_by_policy") def test_require_approval_raises_pending_and_records_request(self): @tide.controlled_tool(name="crm.update_lead", risk_class="internal_write") def update_lead(lead_id): return "updated" with self.assertRaises(tide.LedgerPending) as ctx: update_lead("L-1") approvals = tide_store.read_jsonl(self.project, tide_store.APPROVALS_FILE) self.assertEqual(approvals[0]["type"], "request") self.assertEqual(approvals[0]["approval_id"], ctx.exception.approval_id) self.assertEqual(events(self.project)[0]["approval_status"], "pending") def test_grant_is_hash_bound_and_one_shot(self): @tide.controlled_tool(name="crm.update_lead", risk_class="internal_write") def update_lead(lead_id): return "updated" with self.assertRaises(tide.LedgerPending) as ctx: update_lead("L-1") approval_id = ctx.exception.approval_id tide_store.append_jsonl(self.project, tide_store.APPROVALS_FILE, { "type": "decision", "approval_id": approval_id, "decision": "approved", "approver": "tester", "reason": None, "created_at": tide_store.now_iso(), }) with self.assertRaises(tide.LedgerPending): update_lead("L-2") self.assertEqual(update_lead("L-1"), "updated") ran = [e for e in events(self.project) if e["approval_status"] == "approved"] self.assertEqual(len(ran), 1) with self.assertRaises(tide.LedgerPending): update_lead("L-1") def test_wait_mode_proceeds_after_approval(self): tide.configure(project=self.project, environment="prod", on_approval="wait", approval_timeout=5, approval_poll=0.05) @tide.controlled_tool(name="crm.update_lead", risk_class="internal_write") def update_lead(lead_id): return "updated" def approve_soon(): time.sleep(0.2) reqs = tide_store.read_jsonl(self.project, tide_store.APPROVALS_FILE) tide_store.append_jsonl(self.project, tide_store.APPROVALS_FILE, { "type": "decision", "approval_id": reqs[0]["approval_id"], "decision": "approved", "approver": "tester", "reason": None, "created_at": tide_store.now_iso(), }) t = threading.Thread(target=approve_soon) t.start() self.assertEqual(update_lead("L-1"), "updated") t.join() self.assertEqual(events(self.project)[0]["approval_status"], "approved") def test_wait_mode_denied_raises(self): tide.configure(project=self.project, environment="prod", on_approval="wait", approval_timeout=5, approval_poll=0.05) @tide.controlled_tool(name="crm.update_lead", risk_class="internal_write") def update_lead(lead_id): raise AssertionError("must never run") def deny_soon(): time.sleep(0.2) reqs = tide_store.read_jsonl(self.project, tide_store.APPROVALS_FILE) tide_store.append_jsonl(self.project, tide_store.APPROVALS_FILE, { "type": "decision", "approval_id": reqs[0]["approval_id"], "decision": "denied", "approver": "tester", "reason": "not now", "created_at": tide_store.now_iso(), }) t = threading.Thread(target=deny_soon) t.start() with self.assertRaises(tide.LedgerDenied): update_lead("L-1") t.join() e = events(self.project)[0] self.assertEqual(e["approval_status"], "denied") self.assertEqual(e["error_code"], "approval_denied") def test_outcome_links_to_trace(self): @tide.controlled_tool(name="crm.get_lead", risk_class="read_only") def get_lead(lead_id): return "lead" get_lead("L-1") tide.record_outcome("success", human_corrected=False) outcome = tide_store.read_jsonl(self.project, tide_store.OUTCOMES_FILE)[0] self.assertEqual(outcome["trace_id"], events(self.project)[0]["trace_id"]) self.assertEqual(outcome["status"], "success") def test_packs_generate_reviewable_policy(self): pol = tide_packs.get("support-refunds", environment="prod") self.assertEqual(pol["pack"], "support-refunds") self.assertEqual(pol["tools"]["stripe.refund"]["decision"], "require_approval") self.assertEqual(pol["tools"]["stripe.customer.delete"]["decision"], "block") def test_cli_init_writes_pack_policy(self): out = StringIO() with redirect_stdout(out): code = tide_cli.main(["init", "--project", self.project, "--pack", "sql-analyst", "--force"]) self.assertEqual(code, 0) pol = tide_store.load_policy(self.project) self.assertEqual(pol["pack"], "sql-analyst") self.assertEqual(pol["decisions"]["internal_write"]["prod"], "block") def test_cli_dry_run_does_not_write_policy(self): shutil.rmtree(self.project) out = StringIO() with redirect_stdout(out): code = tide_cli.main(["init", "--project", self.project, "--pack", "mcp-baseline", "--dry-run"]) self.assertEqual(code, 0) self.assertIn('"pack": "mcp-baseline"', out.getvalue()) self.assertFalse(os.path.exists(os.path.join(self.project, tide_store.POLICY_FILE))) def test_cli_approves_pending_request(self): @tide.controlled_tool(name="stripe.refund", risk_class="money_movement", capture=["charge_id"]) def refund(charge_id): return "refunded" with self.assertRaises(tide.LedgerPending) as ctx: refund("ch_1") prefix = ctx.exception.approval_id[:8] out = StringIO() with redirect_stdout(out): code = tide_cli.main(["approve", prefix, "--project", self.project, "--reason", "ok"]) self.assertEqual(code, 0) self.assertIn("one-shot", out.getvalue()) self.assertEqual(tide_store.approval_state(self.project, ctx.exception.approval_id), "approved") def test_cli_report_prints_earned_use(self): @tide.controlled_tool(name="crm.get_lead", risk_class="read_only") def get_lead(lead_id): return "lead" get_lead("L-1") tide.record_outcome("success") out = StringIO() with redirect_stdout(out): code = tide_cli.main(["report", "--project", self.project]) self.assertEqual(code, 0) self.assertIn("Earned use by tool", out.getvalue()) self.assertIn("crm.get_lead", out.getvalue()) def test_cli_scan_generates_wrapper_plan(self): agent = tempfile.mkdtemp(prefix="tide-agent-scan-") self.addCleanup(lambda: shutil.rmtree(agent, ignore_errors=True)) with open(os.path.join(agent, "agent.py"), "w", encoding="utf-8") as f: f.write(textwrap.dedent(""" from langchain.tools import tool @tool def issue_refund(charge_id, amount): return stripe.refunds.create(charge=charge_id, amount=amount) """).lstrip()) plan = os.path.join(agent, ".tide", "tide-plan.json") out = StringIO() with redirect_stdout(out): code = tide_cli.main(["scan", "--path", agent, "--pack", "support-refunds", "--plan", plan]) self.assertEqual(code, 0) with open(plan, "r", encoding="utf-8") as f: data = json.load(f) action = data["actions"][0] self.assertEqual(action["risk_class"], "money_movement") self.assertEqual(action["policy_decision"], "require_approval") self.assertEqual(action["capture"], ["charge_id", "amount"]) self.assertIn("issue_refund", out.getvalue()) def test_cli_apply_adds_wrapper_import_and_policy(self): agent = tempfile.mkdtemp(prefix="tide-agent-apply-") self.addCleanup(lambda: shutil.rmtree(agent, ignore_errors=True)) target = os.path.join(agent, "agent.py") with open(target, "w", encoding="utf-8") as f: f.write(textwrap.dedent(""" def send_customer_email(recipient_email, body): return gmail.send(to=recipient_email, body=body) """).lstrip()) plan = os.path.join(agent, ".tide", "tide-plan.json") with redirect_stdout(StringIO()): self.assertEqual(tide_cli.main(["scan", "--path", agent, "--pack", "support-refunds", "--plan", plan]), 0) self.assertEqual(tide_cli.main(["apply", "--plan", plan, "--yes", "--write-policy"]), 0) with open(target, "r", encoding="utf-8") as f: source = f.read() self.assertIn("from tide import controlled_tool", source) self.assertIn("@controlled_tool(name='gmail.send', risk_class='customer_visible', capture=['recipient_email'])", source) ast.parse(source) pol = tide_store.load_policy(os.path.join(agent, ".tide", "ledger")) self.assertEqual(pol["pack"], "support-refunds") def test_cli_scan_generates_js_wrapper_plan(self): agent = tempfile.mkdtemp(prefix="tide-js-agent-scan-") self.addCleanup(lambda: shutil.rmtree(agent, ignore_errors=True)) with open(os.path.join(agent, "server.mjs"), "w", encoding="utf-8") as f: f.write(textwrap.dedent(""" export function normalizeRefundRequest(input = {}) { return input; } export async function createRefund(input, deps = {}) { const client = deps.client; return client.refunds.create({ payment_intent: input.payment_intent, amount: input.amount_cents, }); } """).lstrip()) plan = os.path.join(agent, ".tide", "tide-plan.json") out = StringIO() with redirect_stdout(out): code = tide_cli.main(["scan", "--path", agent, "--pack", "support-refunds", "--plan", plan]) self.assertEqual(code, 0) with open(plan, "r", encoding="utf-8") as f: data = json.load(f) self.assertEqual(data["summary"]["files_scanned"], 1) self.assertEqual(data["summary"]["proposed"], 1) action = data["actions"][0] self.assertEqual(action["language"], "javascript") self.assertEqual(action["function"], "createRefund") self.assertEqual(action["tool_name"], "stripe.refund") self.assertEqual(action["policy_decision"], "require_approval") self.assertIn("payment_intent", action["capture"]) self.assertIn("createRefund", out.getvalue()) def test_cli_apply_wraps_js_refund_tool(self): agent = tempfile.mkdtemp(prefix="tide-js-agent-apply-") self.addCleanup(lambda: shutil.rmtree(agent, ignore_errors=True)) target = os.path.join(agent, "server.mjs") with open(target, "w", encoding="utf-8") as f: f.write(textwrap.dedent(""" #!/usr/bin/env node export async function createRefund(input, deps = {}) { const client = deps.client; return client.refunds.create({ payment_intent: input.payment_intent, amount: input.amount_cents, }); } """).lstrip()) plan = os.path.join(agent, ".tide", "tide-plan.json") with redirect_stdout(StringIO()): self.assertEqual(tide_cli.main(["scan", "--path", agent, "--pack", "support-refunds", "--plan", plan]), 0) self.assertEqual(tide_cli.main(["apply", "--plan", plan, "--yes", "--write-policy"]), 0) with open(target, "r", encoding="utf-8") as f: source = f.read() self.assertIn('import { controlledTool } from "./.tide/tide-runtime.mjs";', source) self.assertIn("export const createRefund = controlledTool", source) self.assertTrue(os.path.exists(os.path.join(agent, ".tide", "tide-runtime.mjs"))) subprocess.run(["node", "--check", target], check=True, capture_output=True, text=True) pol = tide_store.load_policy(os.path.join(agent, ".tide", "ledger")) self.assertEqual(pol["tools"]["stripe.refund"]["decision"], "require_approval") def test_cli_apply_interactive_sets_policy_decision(self): agent = tempfile.mkdtemp(prefix="tide-agent-interactive-") self.addCleanup(lambda: shutil.rmtree(agent, ignore_errors=True)) target = os.path.join(agent, "agent.py") with open(target, "w", encoding="utf-8") as f: f.write(textwrap.dedent(""" def issue_refund(charge_id, amount): return stripe.refunds.create(charge=charge_id, amount=amount) """).lstrip()) plan = os.path.join(agent, ".tide", "tide-plan.json") out = StringIO() with redirect_stdout(out): self.assertEqual(tide_cli.main(["scan", "--path", agent, "--pack", "support-refunds", "--plan", plan]), 0) with redirect_stdout(out), patch("builtins.input", return_value="block"): self.assertEqual(tide_cli.main(["apply", "--plan", plan, "--interactive", "--write-policy"]), 0) with open(target, "r", encoding="utf-8") as f: source = f.read() self.assertIn("@controlled_tool(name='stripe.refund', risk_class='money_movement', capture=['charge_id', 'amount'])", source) pol = tide_store.load_policy(os.path.join(agent, ".tide", "ledger")) self.assertEqual(pol["tools"]["stripe.refund"]["decision"], "block") self.assertIn("explicit tool decision", out.getvalue()) self.assertIn("selected: block", out.getvalue()) def test_cli_run_scans_applies_and_writes_policy(self): agent = tempfile.mkdtemp(prefix="tide-agent-run-") self.addCleanup(lambda: shutil.rmtree(agent, ignore_errors=True)) target = os.path.join(agent, "agent.py") with open(target, "w", encoding="utf-8") as f: f.write(textwrap.dedent(""" def issue_refund(charge_id, amount): return stripe.refunds.create(charge=charge_id, amount=amount) """).lstrip()) out = StringIO() with redirect_stdout(out), patch("builtins.input", return_value="escalate"): self.assertEqual(tide_cli.main(["run", "--path", agent, "--pack", "support-refunds"]), 0) with open(target, "r", encoding="utf-8") as f: source = f.read() self.assertIn("@controlled_tool(name='stripe.refund', risk_class='money_movement', capture=['charge_id', 'amount'])", source) pol = tide_store.load_policy(os.path.join(agent, ".tide", "ledger")) self.assertEqual(pol["tools"]["stripe.refund"]["decision"], "require_approval") self.assertIn("run your agent normally", out.getvalue()) def test_cli_run_defaults_baseline_and_accepts_manual_decision(self): agent = tempfile.mkdtemp(prefix="tide-agent-run-baseline-") self.addCleanup(lambda: shutil.rmtree(agent, ignore_errors=True)) target = os.path.join(agent, "agent.py") with open(target, "w", encoding="utf-8") as f: f.write(textwrap.dedent(""" def issue_refund(charge_id, amount): return stripe.refunds.create(charge=charge_id, amount=amount) """).lstrip()) out = StringIO() with redirect_stdout(out), patch("builtins.input", return_value="deny") as user_input: self.assertEqual(tide_cli.main(["run", "--path", agent]), 0) user_input.assert_called_once() pol = tide_store.load_policy(os.path.join(agent, ".tide", "ledger")) self.assertEqual(pol["pack"], "baseline") self.assertEqual(pol["tools"]["stripe.refund"]["decision"], "block") self.assertIn("selected: block", out.getvalue()) def test_cli_apply_preserves_framework_decorator_order(self): agent = tempfile.mkdtemp(prefix="tide-agent-order-") self.addCleanup(lambda: shutil.rmtree(agent, ignore_errors=True)) target = os.path.join(agent, "agent.py") with open(target, "w", encoding="utf-8") as f: f.write(textwrap.dedent(""" from langchain.tools import tool @tool def update_deal(deal_id, stage): return hubspot.crm.deals.basic_api.update(deal_id, {"stage": stage}) """).lstrip()) plan = os.path.join(agent, ".tide", "tide-plan.json") with redirect_stdout(StringIO()): self.assertEqual(tide_cli.main(["scan", "--path", agent, "--pack", "baseline", "--plan", plan]), 0) self.assertEqual(tide_cli.main(["apply", "--plan", plan, "--yes"]), 0) with open(target, "r", encoding="utf-8") as f: lines = [line.strip() for line in f.readlines()] self.assertLess(lines.index("@tool"), lines.index("@controlled_tool(name='hubspot.update_deal', risk_class='internal_write', capture=['deal_id'])")) self.assertLess(lines.index("@controlled_tool(name='hubspot.update_deal', risk_class='internal_write', capture=['deal_id'])"), lines.index("def update_deal(deal_id, stage):")) def test_cli_scan_reports_already_wrapped(self): agent = tempfile.mkdtemp(prefix="tide-agent-wrapped-") self.addCleanup(lambda: shutil.rmtree(agent, ignore_errors=True)) with open(os.path.join(agent, "agent.py"), "w", encoding="utf-8") as f: f.write(textwrap.dedent(""" from tide import controlled_tool @controlled_tool(name="stripe.refund", risk_class="money_movement") def issue_refund(charge_id): return stripe.refunds.create(charge=charge_id) """).lstrip()) plan = os.path.join(agent, ".tide", "tide-plan.json") with redirect_stdout(StringIO()): self.assertEqual(tide_cli.main(["scan", "--path", agent, "--plan", plan]), 0) with open(plan, "r", encoding="utf-8") as f: data = json.load(f) self.assertEqual(data["summary"]["proposed"], 0) self.assertEqual(data["summary"]["already_wrapped"], 1) def test_cli_scan_reports_already_wrapped_js(self): agent = tempfile.mkdtemp(prefix="tide-js-agent-wrapped-") self.addCleanup(lambda: shutil.rmtree(agent, ignore_errors=True)) with open(os.path.join(agent, "server.mjs"), "w", encoding="utf-8") as f: f.write(textwrap.dedent(""" import { controlledTool } from "./.tide/tide-runtime.mjs"; export const createRefund = controlledTool({ name: 'stripe.refund', riskClass: 'money_movement' }, async (input) => { return client.refunds.create(input); }); """).lstrip()) plan = os.path.join(agent, ".tide", "tide-plan.json") with redirect_stdout(StringIO()): self.assertEqual(tide_cli.main(["scan", "--path", agent, "--plan", plan]), 0) with open(plan, "r", encoding="utf-8") as f: data = json.load(f) self.assertEqual(data["summary"]["proposed"], 0) self.assertEqual(data["summary"]["already_wrapped"], 1) def test_cli_scan_skips_common_helpers(self): agent = tempfile.mkdtemp(prefix="tide-agent-helpers-") self.addCleanup(lambda: shutil.rmtree(agent, ignore_errors=True)) with open(os.path.join(agent, "agent.py"), "w", encoding="utf-8") as f: f.write(textwrap.dedent(""" def create_sql_agent(db): return object() def summarize_tool_args(args): return str(args) def main(): return create_sql_agent(None) """).lstrip()) plan = os.path.join(agent, ".tide", "tide-plan.json") with redirect_stdout(StringIO()): self.assertEqual(tide_cli.main(["scan", "--path", agent, "--plan", plan]), 0) with open(plan, "r", encoding="utf-8") as f: data = json.load(f) self.assertEqual(data["summary"]["proposed"], 0) if __name__ == "__main__": unittest.main()