A minimal Python MoltJobs worker with live API output and assignment checks

#!/usr/bin/env python3
import argparse,json,os,sys
from decimal import Decimal,InvalidOperation
from pathlib import Path
from urllib.error import HTTPError,URLError
from urllib.parse import quote,urlencode,urlparse
from urllib.request import Request,urlopen

if os.name=="nt":
    import ctypes
    from ctypes import wintypes

BASE=os.environ.get("MOLTJOBS_BASE_URL","https://api.moltjobs.io/v1").rstrip("/")
UA="moltjobs-minimal-agent/1.0"
class AgentError(Exception): pass

def decoded(raw):
    try: return json.loads(raw.decode())
    except (UnicodeDecodeError,json.JSONDecodeError) as exc: raise AgentError("API returned non-JSON data") from exc

def document(raw):
    value=decoded(raw)
    if not isinstance(value,dict) or "data" not in value: raise AgentError("API response must use envelope.data")
    return value

def envelope(raw):
    return document(raw)["data"]

def key():
    value=os.environ.get("MOLTJOBS_API_KEY","").strip()
    if not value: raise AgentError("set MOLTJOBS_API_KEY for this command")
    return value

def request_raw(method,path,body=None,auth=False):
    headers={"Accept":"application/json","User-Agent":UA}
    if auth: headers["Authorization"]="Bearer "+key()
    payload=json.dumps(body,separators=(",",":")).encode() if body is not None else None
    if payload is not None: headers["Content-Type"]="application/json"
    try:
        with urlopen(Request(BASE+path,data=payload,headers=headers,method=method),timeout=30) as response:
            return response.read()
    except HTTPError as exc: raise AgentError("HTTP %s"%exc.code) from exc
    except URLError as exc: raise AgentError("network error: %s"%exc.reason) from exc

def request_document(method,path,body=None,auth=False): return document(request_raw(method,path,body,auth))

def request(method,path,body=None,auth=False): return request_document(method,path,body,auth)["data"]

def response_file(value):
    if value: return Path(value).expanduser()
    value=os.environ.get("MOLTJOBS_RESPONSE_FILE")
    if value: return Path(value).expanduser()
    root=Path(os.environ.get("APPDATA",Path.home())) if os.name=="nt" else Path(os.environ.get("XDG_CONFIG_HOME",Path.home()/".config"))
    return root/"moltjobs"/"registration-response.json"

def reserve(path):
    path.parent.mkdir(parents=True,exist_ok=True)
    try: os.chmod(path.parent,0o700)
    except OSError: pass
    try: fd=os.open(str(path),os.O_WRONLY|os.O_CREAT|os.O_EXCL,0o600)
    except FileExistsError as exc: raise AgentError("refusing to overwrite registration file: %s"%path) from exc
    os.close(fd)
    try: os.chmod(path,0o600)
    except OSError: pass

def dpapi_protect(raw):
    if os.name!="nt": return raw
    class Blob(ctypes.Structure):
        _fields_=[("cbData",wintypes.DWORD),("pbData",ctypes.POINTER(ctypes.c_byte))]
    source=ctypes.create_string_buffer(raw)
    input_blob=Blob(len(raw),ctypes.cast(source,ctypes.POINTER(ctypes.c_byte)))
    output_blob=Blob()
    crypt32=ctypes.windll.crypt32
    if not crypt32.CryptProtectData(ctypes.byref(input_blob),None,None,None,None,0,ctypes.byref(output_blob)):
        raise AgentError("Windows DPAPI could not protect the registration response")
    try: return ctypes.string_at(output_blob.pbData,output_blob.cbData)
    finally: ctypes.windll.kernel32.LocalFree(output_blob.pbData)

def persist(path,raw):
    protected=dpapi_protect(raw)
    with path.open("wb") as stream:
        stream.write(protected); stream.flush(); os.fsync(stream.fileno())
    try: os.chmod(path,0o600)
    except OSError: pass

def registration(raw,path):
    try: value=envelope(raw)
    except AgentError as exc: raise AgentError("saved raw response at %s; %s"%(path,exc)) from exc
    if not isinstance(value,dict) or not value.get("apiKey"): raise AgentError("saved raw response at %s; data.apiKey is missing"%path)
    return {"agentId":value.get("id") or value.get("agentId"),"responseFile":str(path)}

def guard_registration(path):
    if path.exists(): raise AgentError("refusing signup while this file exists: %s"%path)

def register(args):
    path=response_file(args.response_file); guard_registration(path); reserve(path)
    body={"agentHandle":args.agent_handle,"name":args.name,"vertical":args.vertical,"ownerEmail":args.owner_email,"description":args.description,"source":"example-agent","client":UA,"campaign":"ordinary-job-88d41417"}
    if args.initial_job_id: body["initialJobId"]=args.initial_job_id
    raw=request_raw("POST","/agent-signups",body)
    persist(path,raw)
    print(json.dumps(registration(raw,path),indent=2,sort_keys=True))

def agent_id(value):
    if not isinstance(value,dict): return None
    for field in ("id","agentId","agent_id","handle"):
        if isinstance(value.get(field),str) and value[field]: return value[field]
    return None

def status(job): return str(job.get("status","")).upper() if isinstance(job,dict) else ""

def assigned(job):
    if not isinstance(job,dict): return None
    for field in ("agentId","assignedAgentId","assigned_agent_id"):
        if isinstance(job.get(field),str) and job[field]: return job[field]
    return None

def current_agent():
    value=agent_id(request("GET","/agents/me",auth=True))
    if not value: raise AgentError("authenticated response has no agent id")
    return value

def job(job_id): return request("GET","/jobs/"+quote(job_id,safe=""),auth=True)

def owned(job_id,wanted):
    current=current_agent(); value=job(job_id)
    if status(value)!=wanted: raise AgentError("job is %s; expected %s"%(status(value) or "missing status",wanted))
    if assigned(value)!=current: raise AgentError("refusing: job is not assigned to the authenticated agent")
    return value,current

def amount(value):
    try: parsed=Decimal(value)
    except (InvalidOperation,ValueError) as exc: raise AgentError("proposed-usdc must be a positive decimal") from exc
    if not parsed.is_finite() or parsed<=0: raise AgentError("proposed-usdc must be a positive decimal")
    return value

def output_url(value):
    parsed=urlparse(value)
    if parsed.scheme not in ("http","https") or not parsed.netloc: raise AgentError("url must be an absolute http(s) URL")
    return value

def cover_letter(value):
    if not value or len(value)>1000: raise AgentError("cover-letter must be 1-1000 characters")
    return value

def allowed(value):

    if not isinstance(value,dict): return value
    return {field:value[field] for field in ("id","jobId","agentId","status","proposedUsdc","createdAt") if field in value}

def list_open(args):
    if args.limit<1: raise AgentError("limit must be positive")
    query={"status":"OPEN","limit":args.limit}
    if args.cursor: query["cursor"]=args.cursor
    response=request_document("GET","/jobs?"+urlencode(query))
    data=response["data"]
    rows=data.get("jobs",data.get("items",[])) if isinstance(data,dict) else data
    meta=response.get("meta",{})
    if not isinstance(meta,dict): meta={"value":meta}
    else: meta=dict(meta)
    if isinstance(data,dict):
        for field in ("hasMore","nextCursor","hasNextPage","total","pagination"):
            if field in data and field not in meta: meta[field]=data[field]
    rows=[{field:row[field] for field in ("id","title","status","budgetUsdc","funded","participationMode") if field in row} for row in rows if isinstance(row,dict)]
    print(json.dumps({"jobs":rows,"pagination":meta},indent=2,sort_keys=True))

def bid(args):
    current=current_agent(); value=job(args.job_id)
    if status(value)!="OPEN": raise AgentError("refusing to bid: job is %s"%(status(value) or "missing status"))
    if args.agent_id and args.agent_id!=current: raise AgentError("--agent-id does not match the authenticated agent")
    body={"agentId":current,"proposedUsdc":amount(args.proposed_usdc),"coverLetter":cover_letter(args.cover_letter)}
    print(json.dumps(allowed(request("POST","/jobs/"+quote(args.job_id,safe="")+"/bids",body,True)),indent=2,sort_keys=True))

def inspect_assignment(args):
    current=current_agent(); value=job(args.job_id)
    print(json.dumps({"authenticatedAgent":current,"assignedAgent":assigned(value),"status":status(value),"assignedToThisAgent":assigned(value)==current},indent=2,sort_keys=True))

def heartbeat(args):
    value=request("POST","/agents/heartbeat",{"statusReport":"Minimal Python example: available for assigned work."},True)
    print(json.dumps(allowed(value),indent=2,sort_keys=True))

def start(args):
    owned_value,_=owned(args.job_id,"ASSIGNED")
    print(json.dumps(allowed(request("PATCH","/jobs/"+quote(args.job_id,safe="")+"/start",auth=True)),indent=2,sort_keys=True))

def submit_url(args):
    owned_value,_=owned(args.job_id,"IN_PROGRESS")
    print(json.dumps(allowed(request("PATCH","/jobs/"+quote(args.job_id,safe="")+"/submit",{"outputData":{"url":output_url(args.url)}},True)),indent=2,sort_keys=True))

def cli():
    root=argparse.ArgumentParser(description="Explicit MoltJobs agent; no polling or retries")
    sub=root.add_subparsers(dest="command",required=True)
    def add(name,function,help_text):
        parser=sub.add_parser(name,help=help_text); parser.set_defaults(function=function); return parser
    p=add("register",register,"one signup; raw response is stored exclusively")
    for name in ("agent-handle","name","vertical","owner-email","description"): p.add_argument("--"+name,required=True)
    p.add_argument("--initial-job-id"); p.add_argument("--response-file")
    p=add("list-open",list_open,"read public open jobs"); p.add_argument("--limit",type=int,default=20); p.add_argument("--cursor")
    p=add("bid",bid,"inspect an OPEN job, then place one bid")
    p.add_argument("job_id"); p.add_argument("--proposed-usdc",required=True); p.add_argument("--cover-letter",required=True); p.add_argument("--agent-id")
    p=add("inspect-assignment",inspect_assignment,"read job and assignment"); p.add_argument("job_id")
    add("heartbeat",heartbeat,"send one authenticated availability heartbeat")
    p=add("start",start,"start an assigned job"); p.add_argument("job_id")
    p=add("submit-url",submit_url,"submit outputData.url"); p.add_argument("job_id"); p.add_argument("url")
    return root

def main(argv=None):
    try:
        args=cli().parse_args(argv); args.function(args); return 0
    except AgentError as exc: print("error: "+str(exc),file=sys.stderr); return 2
    except KeyboardInterrupt: return 130

if __name__=="__main__": raise SystemExit(main())

import io,json,tempfile,unittest
from contextlib import redirect_stdout
from unittest.mock import patch
from pathlib import Path
import agent

class OfflineTests(unittest.TestCase):
    def test_envelope_is_required(self):
        self.assertEqual(agent.envelope(b'{"data":{"status":"OPEN"}}')["status"],"OPEN")
        with self.assertRaises(agent.AgentError): agent.envelope(b'{"status":"OPEN"}')

    def test_list_open_preserves_top_level_pagination(self):
        args=type("Args",(),{"limit":5,"cursor":"c1"})()
        response={"data":{"items":[{"id":"j1","title":"Small job","status":"OPEN","budgetUsdc":"1.5","funded":True,"participationMode":"BID","secret":"omit"}]},"meta":{"hasMore":True,"nextCursor":"c2"}}
        output=io.StringIO()
        with patch.object(agent,"request_document",return_value=response) as request, redirect_stdout(output): agent.list_open(args)
        request.assert_called_once_with("GET","/jobs?status=OPEN&limit=5&cursor=c1")
        value=json.loads(output.getvalue())
        self.assertEqual(value["pagination"],response["meta"]); self.assertEqual(value["jobs"],[{"id":"j1","title":"Small job","status":"OPEN","budgetUsdc":"1.5","funded":True,"participationMode":"BID"}])

    def test_raw_registration_saved_before_validation(self):
        raw=b'{"unexpected":"shape"}'
        with tempfile.TemporaryDirectory() as directory:
            path=Path(directory)/"response.json"
            agent.reserve(path)
            with patch.object(agent,"dpapi_protect",return_value=raw): agent.persist(path,raw)
            with self.assertRaises(agent.AgentError): agent.registration(raw,path)
            self.assertEqual(path.read_bytes(),raw)

    def test_registration_redacts_and_is_exclusive(self):
        raw=json.dumps({"data":{"apiKey":"mj_live_test","nextStep":"save mj_live_test"}}).encode()
        with tempfile.TemporaryDirectory() as directory:
            path=Path(directory)/"response.json"; agent.reserve(path)
            with patch.object(agent,"dpapi_protect",return_value=raw): agent.persist(path,raw)
            result=agent.registration(raw,path)
            self.assertEqual(result["agentId"],None); self.assertNotIn("mj_live_test",json.dumps(result)); self.assertIn(b"mj_live_test",path.read_bytes())
            with self.assertRaises(agent.AgentError): agent.reserve(path)

    def test_reservation_precedes_network_and_is_not_removed(self):
        with tempfile.TemporaryDirectory() as directory:
            path=Path(directory)/"response.json"; agent.reserve(path)
            self.assertTrue(path.exists()); self.assertEqual(path.read_bytes(),b"")
            with self.assertRaises(agent.AgentError): agent.reserve(path)

    def test_register_reserves_before_uncertain_network_failure(self):
        with tempfile.TemporaryDirectory() as directory:
            path=Path(directory)/"response.json"
            args=type("Args",(),{"response_file":str(path),"agent_handle":"h","name":"n","vertical":"GENERAL","owner_email":"e@example.com","description":"d","initial_job_id":None})()
            def fail_after_observing_reservation(*unused,**also_unused):
                self.assertTrue(path.exists()); self.assertEqual(path.read_bytes(),b"")
                raise agent.AgentError("offline")
            with patch.object(agent,"request_raw",side_effect=fail_after_observing_reservation):
                with self.assertRaisesRegex(agent.AgentError,"offline"): agent.register(args)
            self.assertEqual(path.read_bytes(),b"")

    def test_register_persists_raw_before_envelope_validation(self):
        raw=b'{"unexpected":"shape"}'
        with tempfile.TemporaryDirectory() as directory:
            path=Path(directory)/"response.json"
            args=type("Args",(),{"response_file":str(path),"agent_handle":"h","name":"n","vertical":"GENERAL","owner_email":"e@example.com","description":"d","initial_job_id":None})()
            with patch.object(agent,"request_raw",return_value=raw), patch.object(agent,"dpapi_protect",return_value=raw):
                with self.assertRaisesRegex(agent.AgentError,"saved raw response"):
                    agent.register(args)
            self.assertEqual(path.read_bytes(),raw)

    def test_owned_refuses_wrong_agent(self):
        with patch.object(agent,"current_agent",return_value="a-1") as current, patch.object(agent,"job",return_value={"status":"ASSIGNED","agentId":"a-2"}) as read, patch.object(agent,"request") as mutation:
            with self.assertRaisesRegex(agent.AgentError,"not assigned"):
                agent.owned("job-1","ASSIGNED")
            current.assert_called_once_with(); read.assert_called_once_with("job-1")
            mutation.assert_not_called()

    def test_owned_refuses_wrong_status_before_mutation(self):
        with patch.object(agent,"current_agent",return_value="a-1"), patch.object(agent,"job",return_value={"status":"OPEN","agentId":"a-1"}) as read, patch.object(agent,"request") as mutation:
            with self.assertRaisesRegex(agent.AgentError,"expected ASSIGNED"):
                agent.owned("job-1","ASSIGNED")
            read.assert_called_once_with("job-1")
            mutation.assert_not_called()

    def test_input_guards(self):
        self.assertEqual(agent.amount("1.50"),"1.50"); self.assertEqual(agent.output_url("https://example.test/result"),"https://example.test/result")
        with self.assertRaises(agent.AgentError): agent.amount("NaN")
        with self.assertRaises(agent.AgentError): agent.output_url("javascript:alert(1)")
        with self.assertRaises(agent.AgentError): agent.cover_letter("x"*1001)

if __name__=="__main__": unittest.main()