diff --git a/backend/chainlit/__init__.py b/backend/chainlit/__init__.py index 99dd9d15b2..dbb7ee5389 100644 --- a/backend/chainlit/__init__.py +++ b/backend/chainlit/__init__.py @@ -104,6 +104,7 @@ from chainlit.mistralai import instrument_mistralai from chainlit.openai import instrument_openai from chainlit.semantic_kernel import SemanticKernelFilter + from chainlit.taskmarket import TaskmarketTool def sleep(duration: int): @@ -132,6 +133,7 @@ def acall(self): "instrument_openai": "chainlit.openai", "instrument_mistralai": "chainlit.mistralai", "SemanticKernelFilter": "chainlit.semantic_kernel", + "TaskmarketTool": "chainlit.taskmarket", "server": "chainlit.server", } ) @@ -174,6 +176,7 @@ def acall(self): "Task", "TaskList", "TaskStatus", + "TaskmarketTool", "Text", "User", "Video", diff --git a/backend/chainlit/sample/taskmarket.py b/backend/chainlit/sample/taskmarket.py new file mode 100644 index 0000000000..86ad707f65 --- /dev/null +++ b/backend/chainlit/sample/taskmarket.py @@ -0,0 +1,72 @@ +# Taskmarket requester tool demo for Chainlit. +# +# Run with: chainlit run backend/chainlit/sample/taskmarket.py +# The chat demonstrates the requester flow: it shows live open tasks, issues +# a fresh single-use authorization token, then creates a spend-capped funded +# task through the official Taskmarket CLI with that token. + +import json + +import chainlit as cl +from chainlit.taskmarket import TaskmarketTool + + +@cl.on_chat_start +async def on_chat_start() -> None: + await cl.Message( + content=( + "I can manage **Taskmarket** (onchain agent labor on Base) from " + "this chat: list open tasks, track status, review submissions, " + "and create a funded task -- with a spend cap and a fresh " + "single-use authorization token before anything is created." + ) + ).send() + + +@cl.on_message +async def on_message(message: cl.Message) -> None: + tool = TaskmarketTool() + + list_out = await tool.list_tasks(status="open", limit=3) + await cl.Message(content=f"Open tasks right now:\n```json\n{list_out}\n```").send() + + # Issue a fresh, single-use, time-limited authorization token (2 USDC). + token_payload = json.loads(await tool.request_authorization(reward=2.0)) + token = token_payload["authorizationToken"] + await cl.Message( + content=( + f"Authorization token issued (single-use, expires in " + f"{token_payload['expiresInSeconds']}s): `{token}`" + ) + ).send() + + # User pastes back the exact token. create_task refuses to move money + # without a valid, unused, unexpired token for this reward. + auth = await cl.AskUserMessage( + content="Paste the authorization token to create the funded task " + "(2 USDC, 72h, public).", + timeout=120, + ).send() + + if not auth or not auth.get("output"): + await cl.Message(content="No token -> nothing was created.").send() + return + + result = await tool.create_task( + description=message.content, + reward=2.0, + duration_hours=72, + authorization_token=str(auth["output"]).strip(), + ) + try: + parsed = json.loads(result) + except json.JSONDecodeError: + await cl.Message(content=result).send() + return + + if parsed.get("created") is True: + await cl.Message(content=result).send() + else: + await cl.Message( + content=f"Not created -- reconcile before retrying:\n```json\n{result}\n```" + ).send() diff --git a/backend/chainlit/taskmarket.py b/backend/chainlit/taskmarket.py new file mode 100644 index 0000000000..8a460f6443 --- /dev/null +++ b/backend/chainlit/taskmarket.py @@ -0,0 +1,444 @@ +"""Taskmarket tool for Chainlit apps. + +Taskmarket (https://taskmarket.dev) is an onchain agent labor marketplace on +Base. This module gives Chainlit apps a chat-native Taskmarket requester +tool: discover open tasks, track live status, review submissions, and (with +explicit authorization and a spend cap) create a funded task. + +Every method surfaces as a Chainlit ``Step(type="tool")`` so tool activity is +visible in the chat UI. Reads hit the public Taskmarket REST API through an +async HTTP client and need no wallet. The funded write path shells out to the +official ``taskmarket`` CLI through an async subprocess so wallet keys, the +X402 USDC payment, legal acceptance, and idempotency are handled by +first-party tooling -- this module never touches private keys, seed phrases, +or tokens. +""" + +import asyncio +import json +import os +import secrets +import shutil +import time +import urllib.parse +from typing import Any, Optional + +import httpx + +from chainlit.step import step + +TASKMARKET_API_BASE = "https://api.taskmarket.dev/api" +DEFAULT_MAX_SPEND = float(os.environ.get("TASKMARKET_MAX_SPEND", "5.0")) +DEFAULT_AUTH_TTL_SECONDS = 300 +MAX_AUTH_TOKENS = 100 +MAX_AUTH_TOKENS_PER_SCOPE = 10 +HTTP_TIMEOUT_SECONDS = 45.0 +CLI_TIMEOUT_SECONDS = 120 + + +async def _get_json(url: str, timeout: float = HTTP_TIMEOUT_SECONDS) -> Any: + """Fetch a JSON body over async HTTP (never blocks the event loop).""" + async with httpx.AsyncClient(timeout=timeout) as client: + resp = await client.get(url) + resp.raise_for_status() + return resp.json() + + +def _unknown_result(message: str, cli_output: Optional[str] = None) -> str: + """Build an explicit unknown-settlement result (never raises the ID away).""" + payload: dict[str, Any] = { + "created": "unknown", + "taskId": None, + "message": message, + } + if cli_output is not None: + payload["cliOutput"] = cli_output + return json.dumps(payload, indent=2) + + +class TaskmarketTool: + """Chainlit tool facade for the Taskmarket requester flow. + + Args: + api_base: Taskmarket REST base URL. + max_spend: Hard cap (USDC) for any single funded task creation. + cli_path: Path/name of the official ``taskmarket`` CLI. + auth_ttl_seconds: Lifetime of a generated authorization token. + """ + + def __init__( + self, + api_base: str = TASKMARKET_API_BASE, + max_spend: float = DEFAULT_MAX_SPEND, + cli_path: str = "taskmarket", + auth_ttl_seconds: int = DEFAULT_AUTH_TTL_SECONDS, + ) -> None: + self.api_base = api_base.rstrip("/") + self.max_spend = max_spend + self.cli_path = cli_path + self.auth_ttl_seconds = auth_ttl_seconds + # token -> {"reward": float, "expires_at": float (monotonic), + # "used": bool, "scope": str (conversation thread id, or + # "no-context" when no Chainlit context is available)} + self._auth_tokens: dict[str, dict[str, Any]] = {} + + @staticmethod + def _thread_id() -> Optional[str]: + """Best-effort conversation thread id for token binding. + + Unlike the raw session id, a thread id survives across requests: + websocket sessions keep their thread across reconnections, and + HTTP API callers pass the same ``threadId`` across requests. + Binding to it keeps tokens scoped to one conversation without + breaking the HTTP flow, where the session id changes per request. + """ + try: + from chainlit.context import context + + return getattr(getattr(context, "session", None), "thread_id", None) + except Exception: + return None + + @step(name="taskmarket_request_authorization", type="tool") + async def request_authorization( + self, reward: float, ttl_seconds: Optional[int] = None + ) -> str: + """Generate a fresh, time-limited, single-use authorization token. + + The returned token is server-generated (random, unguessable), expires + after ``ttl_seconds`` (default ``auth_ttl_seconds``), is bound to the + exact reward, and can be used exactly once. Present it to the human + operator; ``create_task`` refuses to move money without it. + + Args: + reward (float): The exact USDC reward the token authorizes. + ttl_seconds (int): Optional override for the token lifetime. + """ + if reward <= 0: + return "REFUSED: reward must be > 0 USDC." + ttl = self.auth_ttl_seconds if ttl_seconds is None else ttl_seconds + if ttl <= 0: + return "REFUSED: authorization TTL must be > 0 seconds." + # Prune expired/used tokens before adding, so a long-lived tool + # instance never accumulates an unbounded authorization registry. + now = time.monotonic() + for tok, rec in list(self._auth_tokens.items()): + if rec["used"] or rec["expires_at"] <= now: + del self._auth_tokens[tok] + scope = self._thread_id() + if scope is None: + scope = "no-context" + outstanding_for_scope = sum( + 1 for rec in self._auth_tokens.values() if rec["scope"] == scope + ) + if outstanding_for_scope >= MAX_AUTH_TOKENS_PER_SCOPE: + return ( + f"REFUSED: this conversation already has " + f"{outstanding_for_scope} outstanding authorization tokens. " + "Wait for existing tokens to expire or use them before " + "requesting another." + ) + if len(self._auth_tokens) >= MAX_AUTH_TOKENS: + return ( + f"REFUSED: too many outstanding authorization tokens " + f"({len(self._auth_tokens)}). Wait for existing tokens to " + "expire or use them before requesting another." + ) + token = "tm-" + secrets.token_urlsafe(24) + self._auth_tokens[token] = { + "reward": reward, + "expires_at": now + ttl, + "used": False, + "scope": scope, + } + return json.dumps( + { + "authorizationToken": token, + "rewardUSDC": reward, + "expiresInSeconds": ttl, + "note": ( + "Hand this token to the human operator. It is single-use " + "and expires; create_task refuses without it." + ), + }, + indent=2, + ) + + @step(name="taskmarket_list_tasks", type="tool") + async def list_tasks( + self, + status: str = "open", + phase: Optional[str] = None, + mode: Optional[str] = None, + limit: int = 25, + ) -> str: + """List Taskmarket tasks (default: open tasks, newest first). + + Args: + status (str): Filter by status, e.g. "open" (default). + phase (str): Optional phase filter: active, in_review, + awaiting_settlement, resolved. + mode (str): Optional mode filter: bounty, claim, pitch, + benchmark, auction. + limit (int): Maximum number of tasks to return (default 25). + """ + params = {"status": status, "take": str(min(max(limit, 1), 100))} + if phase: + params["phase"] = phase + if mode: + params["mode"] = mode + qs = "&".join(f"{k}={urllib.parse.quote(str(v))}" for k, v in params.items()) + data = await _get_json(f"{self.api_base}/tasks?{qs}") + tasks = data.get("tasks", []) + rows = [] + for t in tasks[:limit]: + reward = t.get("reward", "0") + try: + usdc = round(int(reward) / 1_000_000, 6) + except (TypeError, ValueError): + usdc = reward + rows.append( + { + "id": t.get("id"), + "title": (t.get("description") or "")[:120], + "rewardUSDC": usdc, + "status": t.get("status"), + "phase": t.get("phase"), + "submissionWindowOpen": t.get("submissionWindowOpen"), + "expiryTime": t.get("expiryTime"), + } + ) + return json.dumps(rows, indent=2) + + @step(name="taskmarket_get_task", type="tool") + async def get_task(self, task_id: str) -> str: + """Get the full live status of one Taskmarket task. + + Args: + task_id (str): The full 64-hex Taskmarket task id + (e.g. 0x...). Fetch it from list_tasks if unsure. + """ + data = await _get_json(f"{self.api_base}/tasks/{task_id}") + reward = data.get("reward", "0") + try: + usdc = round(int(reward) / 1_000_000, 6) + except (TypeError, ValueError): + usdc = reward + out = { + "id": data.get("id"), + "status": data.get("status"), + "phase": data.get("phase"), + "mode": data.get("mode"), + "rewardUSDC": usdc, + "awardCount": data.get("awardCount"), + "submissionCount": data.get("submissionCount"), + "submissionWindowOpen": data.get("submissionWindowOpen"), + "expiryTime": data.get("expiryTime"), + } + return json.dumps(out, indent=2) + + @step(name="taskmarket_list_submissions", type="tool") + async def list_submissions(self, task_id: str) -> str: + """List submissions for a Taskmarket task for human review. + + Read-only: this tool NEVER accepts or rejects any submission. A + human requester decides via the official tooling. + + Args: + task_id (str): The full 64-hex Taskmarket task id. + """ + data = await _get_json(f"{self.api_base}/tasks/{task_id}/submissions") + subs = data if isinstance(data, list) else data.get("submissions", []) + rows = [] + for s in subs: + rows.append( + { + "id": s.get("id"), + "workerAddress": s.get("workerAddress"), + "workerAgentId": s.get("workerAgentId"), + "submittedAt": s.get("submittedAt"), + "rejectedAt": s.get("rejectedAt"), + "deliverableHash": s.get("deliverableHash"), + "submitTxHash": s.get("submitTxHash"), + "fileUrl": s.get("fileUrl"), + } + ) + return json.dumps(rows, indent=2) + + @step(name="taskmarket_create_task", type="tool") + async def create_task( + self, + description: str, + reward: float, + duration_hours: int, + mode: str = "bounty", + authorization_token: Optional[str] = None, + task_visibility: str = "public", + ) -> str: + """Create a funded Taskmarket task through the official CLI. + + SAFETY CONTRACT (enforced here, before any money moves): + - An authorization token generated by ``request_authorization`` is + required. Tokens are server-generated, time-limited, single-use, + and bound to the exact reward, so a caller cannot construct the + gate itself. + - The exact cost (reward + platform fees, in USDC on Base) is + computed and SURFACED below; if it exceeds ``self.max_spend`` the + call refuses before invoking anything. + - The actual transfer is delegated to the first-party ``taskmarket`` + CLI through an async subprocess (wallet keys, X402 payment, legal + acceptance, and idempotency are the CLI's responsibility). This + tool never stores or logs private keys, seeds, or tokens. + - Result handling surfaces explicit unknown-settlement states instead + of raising or blind-retrying a payment whose status is unknown. + + Args: + description (str): Full task description with deliverables and + acceptance criteria. + reward (float): Reward in USDC (e.g. 5.0 for 5 USDC). + duration_hours (int): Task duration in hours. + mode (str): Task mode: bounty (default), claim, pitch, benchmark, + auction. + authorization_token (str): Single-use token from + ``request_authorization`` for the exact reward. + task_visibility (str): public (default), unlisted, or private. + """ + if not authorization_token: + return ( + "REFUSED: no authorization token. Call request_authorization " + "first -- it issues a fresh, single-use, time-limited token " + "bound to the exact reward. Nothing was created." + ) + record = self._auth_tokens.get(authorization_token) + if record is None: + return ( + "REFUSED: unknown authorization token. Token was not issued " + "by request_authorization (or the tool was re-created). " + "Nothing was created." + ) + current_scope = self._thread_id() + if current_scope is None: + current_scope = "no-context" + if record.get("scope") != current_scope: + return ( + "REFUSED: authorization token was issued in a different " + "conversation and cannot be used here. Request a fresh " + "token from this conversation. Nothing was created." + ) + if record["used"]: + return ( + "REFUSED: authorization token already used exactly once. " + "Request a fresh token. Nothing was created." + ) + if time.monotonic() > record["expires_at"]: + return ( + "REFUSED: authorization token expired. Request a fresh " + "token. Nothing was created." + ) + if record["reward"] != reward: + return ( + "REFUSED: authorization token is bound to reward " + f"{record['reward']} USDC, not {reward}. Nothing was created." + ) + if reward <= 0: + return "REFUSED: reward must be > 0 USDC. Nothing was created." + if reward > self.max_spend: + return ( + f"REFUSED: reward {reward} USDC exceeds max_spend " + f"{self.max_spend} USDC (set TASKMARKET_MAX_SPEND to raise). " + "Nothing was created." + ) + if duration_hours <= 0: + return "REFUSED: duration_hours must be > 0. Nothing was created." + if not shutil.which(self.cli_path): + return ( + "REFUSED: 'taskmarket' CLI not found on PATH. Install it via " + "npm (official package) to create funded tasks. Nothing was " + "created." + ) + + # Token passes every gate: burn it now so it cannot be reused. + self._auth_tokens[authorization_token]["used"] = True + + cmd = [ + self.cli_path, + "task", + "create", + "--description", + description, + "--reward", + str(reward), + "--duration", + str(duration_hours), + "--mode", + mode, + "--task-visibility", + task_visibility, + ] + proc = None + try: + proc = await asyncio.create_subprocess_exec( + *cmd, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + stdout_b, stderr_b = await asyncio.wait_for( + proc.communicate(), timeout=CLI_TIMEOUT_SECONDS + ) + except asyncio.TimeoutError: + # Terminate and reap the hung subprocess before returning, so + # timed-out CLI commands cannot accumulate or keep running + # unattended after the create command was issued. + if proc is not None and proc.returncode is None: + proc.kill() + await proc.wait() + return _unknown_result( + "CLI exceeded the timeout after the create command was " + "issued. Settlement status is UNKNOWN -- reconcile through " + "'taskmarket inbox' / task get before any retry. Never " + "blindly retry a payment." + ) + out = stdout_b.decode("utf-8", "replace").strip() + err = stderr_b.decode("utf-8", "replace").strip() + if proc.returncode != 0: + return ( + f"taskmarket task create failed (exit {proc.returncode}): {err or out}" + ) + + task_id = self._extract_task_id(out) + if task_id is None: + return _unknown_result( + "CLI exited 0 but no task id was found in its output. " + "Settlement status is UNKNOWN -- reconcile through " + "'taskmarket inbox' before any retry.", + cli_output=out, + ) + + live_status = None + status_unavailable = False + try: + live_status = json.loads(await self.get_task(task_id)) + except Exception: + status_unavailable = True + return json.dumps( + { + "created": True, + "taskId": task_id, + "taskUrl": ( + f"https://taskmarket.dev/tasks/{task_id}" if task_id else None + ), + "cliOutput": out, + "liveStatus": live_status, + "statusUnavailable": status_unavailable, + }, + indent=2, + ) + + @staticmethod + def _extract_task_id(cli_output: str) -> Optional[str]: + """Pull the 64-hex task id out of CLI output, if present.""" + for token in cli_output.replace("\n", " ").split(): + t = token.strip(".,:;'\"") + if t.startswith("0x") and len(t) == 66: + return t + return None diff --git a/backend/tests/test_taskmarket.py b/backend/tests/test_taskmarket.py new file mode 100644 index 0000000000..9c8539fa85 --- /dev/null +++ b/backend/tests/test_taskmarket.py @@ -0,0 +1,472 @@ +import asyncio +import json + +import pytest + +from chainlit.taskmarket import TaskmarketTool + +FAKE_TASK_ID = "0x" + "ab" * 32 + +FIXTURE_TASK = { + "id": FAKE_TASK_ID, + "status": "open", + "phase": "active", + "mode": "bounty", + "reward": "5000000", + "awardCount": 0, + "submissionCount": 1, + "submissionWindowOpen": True, + "expiryTime": "2026-08-25T00:00:00.000Z", +} + +FIXTURE_TASKS = { + "tasks": [ + dict(FIXTURE_TASK, description="Task one", reward="2000000"), + dict( + FIXTURE_TASK, id="0x" + "cd" * 32, description="Task two", reward="4000000" + ), + ] +} + +FIXTURE_SUBMISSIONS = [ + { + "id": "sub-1", + "workerAddress": "0x1111", + "workerAgentId": "agent-1", + "submittedAt": "2026-08-22T00:00:00.000Z", + "rejectedAt": None, + "deliverableHash": "0xhash", + "submitTxHash": "0xtx", + "fileUrl": None, + } +] + + +@pytest.fixture +def mock_get_json(monkeypatch): + async def fake_get_json(url, timeout=45.0): + if url.endswith("/submissions"): + return FIXTURE_SUBMISSIONS + if url.endswith("/tasks") or "?status=" in url: + return FIXTURE_TASKS + return FIXTURE_TASK + + monkeypatch.setattr("chainlit.taskmarket._get_json", fake_get_json) + return fake_get_json + + +@pytest.fixture +def mock_taskmarket_cli(monkeypatch): + """Pretend the taskmarket CLI binary exists so the settlement path is + exercised even on machines/CI without the real CLI installed.""" + monkeypatch.setattr( + "chainlit.taskmarket.shutil.which", lambda _name: "/usr/bin/taskmarket" + ) + + +def test_instantiation(): + tool = TaskmarketTool() + assert tool.max_spend > 0 + assert tool.api_base == "https://api.taskmarket.dev/api" + + +@pytest.mark.asyncio +async def test_list_tasks_mocked(mock_chainlit_context, mock_get_json): + async with mock_chainlit_context: + tool = TaskmarketTool() + out = await tool.list_tasks(status="open", limit=2) + tasks = json.loads(out) + assert isinstance(tasks, list) + assert len(tasks) == 2 + assert tasks[0]["id"] == FAKE_TASK_ID + assert tasks[0]["rewardUSDC"] == 2.0 + + +@pytest.mark.asyncio +async def test_get_task_mocked(mock_chainlit_context, mock_get_json): + async with mock_chainlit_context: + tool = TaskmarketTool() + detail = json.loads(await tool.get_task(FAKE_TASK_ID)) + assert detail["status"] == "open" + assert detail["rewardUSDC"] == 5.0 + + +@pytest.mark.asyncio +async def test_list_submissions_is_read_only(mock_chainlit_context, mock_get_json): + async with mock_chainlit_context: + tool = TaskmarketTool() + subs = json.loads(await tool.list_submissions(FAKE_TASK_ID)) + assert len(subs) == 1 + assert subs[0]["id"] == "sub-1" + assert subs[0]["rejectedAt"] is None + + +@pytest.mark.asyncio +async def test_create_task_refuses_without_token(mock_chainlit_context): + async with mock_chainlit_context: + tool = TaskmarketTool() + out = await tool.create_task(description="test", reward=1.0, duration_hours=24) + assert out.startswith("REFUSED") + assert "request_authorization" in out + assert "Nothing was created" in out + + +@pytest.mark.asyncio +async def test_create_task_refuses_unknown_token(mock_chainlit_context): + async with mock_chainlit_context: + tool = TaskmarketTool() + out = await tool.create_task( + description="test", + reward=1.0, + duration_hours=24, + authorization_token="tm-forged", + ) + assert out.startswith("REFUSED") + assert "unknown authorization token" in out + + +@pytest.mark.asyncio +async def test_token_lifecycle_single_use( + mock_chainlit_context, mock_taskmarket_cli, monkeypatch +): + async with mock_chainlit_context: + tool = TaskmarketTool() + token_payload = json.loads(await tool.request_authorization(reward=1.0)) + token = token_payload["authorizationToken"] + assert token.startswith("tm-") + + # Mock the CLI subprocess so no real money moves. + class FakeProc: + returncode = 0 + + async def communicate(self): + return ( + f"created task {FAKE_TASK_ID} ok".encode(), + b"", + ) + + async def fake_spawn(*args, **kwargs): + return FakeProc() + + async def fake_get_json(url, timeout=45.0): + return FIXTURE_TASK + + monkeypatch.setattr( + "chainlit.taskmarket.asyncio.create_subprocess_exec", fake_spawn + ) + monkeypatch.setattr("chainlit.taskmarket._get_json", fake_get_json) + + first = json.loads( + await tool.create_task( + description="test task", + reward=1.0, + duration_hours=24, + authorization_token=token, + ) + ) + assert first["created"] is True + assert first["taskId"] == FAKE_TASK_ID + + # Single-use: a second execution must refuse. + second = await tool.create_task( + description="test task", + reward=1.0, + duration_hours=24, + authorization_token=token, + ) + assert second.startswith("REFUSED") + assert "already used" in second + + +@pytest.mark.asyncio +async def test_create_task_refuses_expired_token(mock_chainlit_context): + async with mock_chainlit_context: + tool = TaskmarketTool() + token = json.loads(await tool.request_authorization(reward=1.0))[ + "authorizationToken" + ] + tool._auth_tokens[token]["expires_at"] = 0 # expire immediately + out = await tool.create_task( + description="test", + reward=1.0, + duration_hours=24, + authorization_token=token, + ) + assert out.startswith("REFUSED") + assert "expired" in out + + +@pytest.mark.asyncio +async def test_create_task_refuses_reward_mismatch(mock_chainlit_context): + async with mock_chainlit_context: + tool = TaskmarketTool() + token = json.loads(await tool.request_authorization(reward=1.0))[ + "authorizationToken" + ] + out = await tool.create_task( + description="test", + reward=2.0, + duration_hours=24, + authorization_token=token, + ) + assert out.startswith("REFUSED") + assert "bound to reward" in out + + +@pytest.mark.asyncio +async def test_create_task_unknown_when_cli_times_out( + mock_chainlit_context, mock_taskmarket_cli, monkeypatch +): + async with mock_chainlit_context: + tool = TaskmarketTool() + token = json.loads(await tool.request_authorization(reward=1.0))[ + "authorizationToken" + ] + + hung_proc = {} + + class HungProc: + returncode = None + killed = False + + async def communicate(self): + await asyncio.sleep(999) + + def kill(self): + self.killed = True + self.returncode = -9 + + async def wait(self): + return self.returncode + + async def fake_spawn(*args, **kwargs): + hung_proc["proc"] = HungProc() + return hung_proc["proc"] + + async def fake_wait_for(coro, timeout): + # Simulate the wait_for timeout WITHOUT awaiting the hung coro. + coro.close() + raise asyncio.TimeoutError + + monkeypatch.setattr( + "chainlit.taskmarket.asyncio.create_subprocess_exec", fake_spawn + ) + monkeypatch.setattr("chainlit.taskmarket.asyncio.wait_for", fake_wait_for) + + out = json.loads( + await tool.create_task( + description="test", + reward=1.0, + duration_hours=24, + authorization_token=token, + ) + ) + assert out["created"] == "unknown" + assert "UNKNOWN" in out["message"] + # P1: a timed-out CLI process must be killed and reaped, not leaked. + assert hung_proc["proc"].killed is True + assert hung_proc["proc"].returncode == -9 + + +@pytest.mark.asyncio +async def test_create_task_unknown_when_output_unrecognized( + mock_chainlit_context, mock_taskmarket_cli, monkeypatch +): + async with mock_chainlit_context: + tool = TaskmarketTool() + token = json.loads(await tool.request_authorization(reward=1.0))[ + "authorizationToken" + ] + + class FakeProc: + returncode = 0 + + async def communicate(self): + return b"all done, no id here", b"" + + async def fake_spawn(*args, **kwargs): + return FakeProc() + + monkeypatch.setattr( + "chainlit.taskmarket.asyncio.create_subprocess_exec", fake_spawn + ) + + out = json.loads( + await tool.create_task( + description="test", + reward=1.0, + duration_hours=24, + authorization_token=token, + ) + ) + assert out["created"] == "unknown" + assert out["cliOutput"] == "all done, no id here" + + +@pytest.mark.asyncio +async def test_create_task_returns_id_when_status_lookup_fails( + mock_chainlit_context, mock_taskmarket_cli, monkeypatch +): + async with mock_chainlit_context: + tool = TaskmarketTool() + token = json.loads(await tool.request_authorization(reward=1.0))[ + "authorizationToken" + ] + + class FakeProc: + returncode = 0 + + async def communicate(self): + return f"created task {FAKE_TASK_ID} ok".encode(), b"" + + async def fake_spawn(*args, **kwargs): + return FakeProc() + + async def fake_get_json(url, timeout=45.0): + raise RuntimeError("status endpoint down") + + monkeypatch.setattr( + "chainlit.taskmarket.asyncio.create_subprocess_exec", fake_spawn + ) + monkeypatch.setattr("chainlit.taskmarket._get_json", fake_get_json) + + out = json.loads( + await tool.create_task( + description="test", + reward=1.0, + duration_hours=24, + authorization_token=token, + ) + ) + assert out["created"] is True + assert out["taskId"] == FAKE_TASK_ID + assert out["statusUnavailable"] is True + + +@pytest.mark.asyncio +async def test_create_task_refuses_token_issued_in_other_conversation( + mock_chainlit_context, mock_taskmarket_cli, monkeypatch +): + async with mock_chainlit_context: + tool = TaskmarketTool() + token = json.loads(await tool.request_authorization(reward=1.0))[ + "authorizationToken" + ] + # Token was issued in thread "test_thread_id"; bind it to a + # different conversation thread and confirm create_task refuses. + tool._auth_tokens[token]["scope"] = "some-other-thread" + + class FakeProc: + returncode = 0 + + async def communicate(self): + return f"created task {FAKE_TASK_ID} ok".encode(), b"" + + async def fake_spawn(*args, **kwargs): + return FakeProc() + + monkeypatch.setattr( + "chainlit.taskmarket.asyncio.create_subprocess_exec", fake_spawn + ) + + out = await tool.create_task( + description="test", + reward=1.0, + duration_hours=24, + authorization_token=token, + ) + assert out.startswith("REFUSED") + assert "different" in out + assert "conversation" in out + + +@pytest.mark.asyncio +async def test_token_bound_to_thread_survives_session_id_change( + mock_chainlit_context, mock_taskmarket_cli, monkeypatch +): + """HTTP API flow: each request gets a NEW session id but the SAME + thread id, so a token issued in request 1 must still be usable in + request 2. The binding must follow the thread, not the session id.""" + async with mock_chainlit_context as ctx: + tool = TaskmarketTool() + token = json.loads(await tool.request_authorization(reward=1.0))[ + "authorizationToken" + ] + assert tool._auth_tokens[token]["scope"] == "test_thread_id" + # Simulate a follow-up HTTP request: session.id changes, thread_id + # stays put. (The mock session's thread_id attribute is what + # _thread_id() reads; the raw session id is no longer consulted.) + ctx.session.id = "some-other-ephemeral-session-id" + + class FakeProc: + returncode = 0 + + async def communicate(self): + return f"created task {FAKE_TASK_ID} ok".encode(), b"" + + async def fake_spawn(*args, **kwargs): + return FakeProc() + + async def fake_get_json(url, timeout=45.0): + return FIXTURE_TASK + + monkeypatch.setattr( + "chainlit.taskmarket.asyncio.create_subprocess_exec", fake_spawn + ) + monkeypatch.setattr("chainlit.taskmarket._get_json", fake_get_json) + + out = json.loads( + await tool.create_task( + description="test", + reward=1.0, + duration_hours=24, + authorization_token=token, + ) + ) + assert out["created"] is True + assert out["taskId"] == FAKE_TASK_ID + + +@pytest.mark.asyncio +async def test_request_authorization_per_conversation_cap( + mock_chainlit_context, mock_taskmarket_cli, monkeypatch +): + """One conversation must not be able to exhaust the shared token cap + and starve every other conversation: per-scope cap applies first.""" + from chainlit.taskmarket import MAX_AUTH_TOKENS, MAX_AUTH_TOKENS_PER_SCOPE + + assert MAX_AUTH_TOKENS_PER_SCOPE < MAX_AUTH_TOKENS + async with mock_chainlit_context as ctx: + tool = TaskmarketTool() + for _ in range(MAX_AUTH_TOKENS_PER_SCOPE): + out = await tool.request_authorization(reward=1.0) + assert "REFUSED" not in out + # The (N+1)-th token for THIS conversation is refused... + out = await tool.request_authorization(reward=1.0) + assert out.startswith("REFUSED") + assert "this conversation" in out + # ...but a different conversation (different thread) can still + # obtain tokens while the first one is capped. + ctx.session.thread_id = "another-conversation" + out = await tool.request_authorization(reward=1.0) + assert "REFUSED" not in out + + +@pytest.mark.asyncio +async def test_request_authorization_rejects_nonpositive_ttl(mock_chainlit_context): + async with mock_chainlit_context: + tool = TaskmarketTool() + out = await tool.request_authorization(reward=1.0, ttl_seconds=0) + assert out.startswith("REFUSED") + assert "TTL" in out + # None still falls back to the default lifetime. + default = json.loads(await tool.request_authorization(reward=1.0)) + assert default["expiresInSeconds"] == tool.auth_ttl_seconds + + +def test_extract_task_id(): + assert ( + TaskmarketTool._extract_task_id(f"created task {FAKE_TASK_ID} ok") + == FAKE_TASK_ID + ) + assert TaskmarketTool._extract_task_id("no id here") is None