"""Standalone wire checks over the caller's existing CommonsClient. NO JUDGMENT WHATSOEVER. No runtime import, identity, storage, retry, URL fetch, execution or publication-rights grant. Descriptors and raw envelopes contain public contribution bodies: keep them in the caller's existing protected custody, never ordinary logs. Only validate_response's allowlisted reference is body-free. Terms consent asserts publication; it does not establish ownership or delegation. binding is caller-verified /me context (agentId, workspaceId, projectId); this module checks scope consistency, not authentication. send_once delegates exactly one request to CommonsClient, which checks the real HTTP status before returning a dict (it does not return status separately), then validates the response into a body-free reference. No other success status is accepted. To retain known linkage, a caller may pin a descriptor's initially None resourceId (create only) or opportunityId (proposal detail/action only) to a previously verified reference. Derived path bindings cannot be replaced. Do not learn pins from an unvalidated response. Revalidate persisted descriptors on every use. """ from __future__ import annotations import base64 import copy import datetime as dt import hashlib import hmac import json import re import unicodedata from urllib.parse import urlsplit BASE = "/api/matm/commons/work" PROPOSALS = "/api/matm/commons/work-proposals" RESOLUTION = "/api/matm/commons/work-resolution" RESOLUTION_SCHEMA = "concresca.work_resolution.v1" RESOLUTION_REFERENCE_SCHEMA = "concresca.work_resolution_result_reference.v1" OPPORTUNITY_SCHEMA = "concresca.work_opportunity.v1" PROPOSAL_SCHEMA = "concresca.work_proposal.v1" ACTION_SCHEMA = "concresca.work_action.v1" WITHDRAWAL_SCHEMA = "concresca.work_opportunity_withdrawal.v1" PAGE_SCHEMA = "concresca.work_page.v1" REFERENCE_SCHEMA = "concresca.work_result_reference.v1" READ_CHECKPOINT_SCHEMA = "concresca.work_read_checkpoint.v1" READ_OBSERVATION_SCHEMA = "concresca.work_read_observation.v1" MAX_READ_CHECKPOINT_BYTES = 8192 _OP = r"commonswork-[0-9a-f]{24}" _PR = r"commonsproposal-[0-9a-f]{24}" _SEGMENT = r"[A-Za-z0-9][A-Za-z0-9._~-]{0,190}" _SHA = r"[0-9a-f]{64}" _CURSOR = r"cw1\.[A-Za-z0-9_-]+\.[0-9a-f]{64}" _ACTIONS = {"counter", "accept", "refuse", "cancel", "deliver", "decide", "withdraw_delivery", "withdraw_terms", "withdraw_decision"} _CLEANUP = {"refuse", "cancel", "withdraw_delivery", "withdraw_terms", "withdraw_decision"} _PROPOSAL_STATES = {"proposed", "committed", "refused", "cancelled", "submitted", "accepted", "disputed", "withdrawn", "delivery_withdrawn"} _NEUTRAL = {"judgmentState": "NONE", "participantImplication": "NONE", "paymentAuthorized": False, "executionAuthorized": False, "acceptanceProvesTruth": False, "memoryPromotion": False, "historicPayloadsIncluded": False} class WorkClientError(ValueError): """Fixed, body-free error; no repr/str of caller objects is evaluated.""" def __init__(self): super().__init__("work_wire_invalid; details redacted") def _require(condition): if not condition: raise WorkClientError() def _match(value, pattern): return type(value) is str and re.fullmatch(pattern, value) is not None def _fields(value, required, optional=()): _require(type(value) is dict and required <= value.keys() and value.keys() <= required | set(optional)) def _integer(value, low=1, high=2147483646): _require(type(value) is int and low <= value <= high) def _choice(value, choices): _require(type(value) is str and value in choices) def _text(value, maximum): _require(type(value) is str and 1 <= len(value) <= maximum and value == value.strip() and not any(unicodedata.category(c).startswith("C") for c in value)) def _array(value, low, high): _require(type(value) is list and low <= len(value) <= high) def _canonical(value, maximum=1048576): # Reject custom objects, cycles, excessive nesting/nodes and non-finite values # before JSON can call custom conversion methods or emit their representations. pending, nodes, size = [(value, 0)], 0, 0 while pending: item, depth = pending.pop() nodes += 1 _require(depth <= 16 and nodes <= 40000) kind = type(item) _require(kind in (dict, list, str, int, bool, type(None))) if kind is dict: _require(len(item) <= 256 and all(type(k) is str for k in item)) pending.extend((v, depth + 1) for v in item.values()) pending.extend((k, depth + 1) for k in item) elif kind is list: _require(len(item) <= 256) pending.extend((v, depth + 1) for v in item) elif kind is str: size += len(item) _require(size <= maximum) elif kind is int: _require(abs(item) <= 2**63 - 1) try: raw = json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False).encode("utf-8") except (ValueError, UnicodeError, RecursionError): raise WorkClientError() from None _require(len(raw) <= maximum) return raw def _digest(value): return hashlib.sha256(_canonical(value)).hexdigest() def _time(value): _require(_match(value, r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z")) try: return dt.datetime.strptime(value, "%Y-%m-%dT%H:%M:%SZ") except ValueError: raise WorkClientError() from None def _uri(value): _text(value, 1024) _require(not any(c.isspace() for c in value) and "\\" not in value) try: parsed = urlsplit(value) _require(parsed.scheme == "https" and bool(parsed.hostname) and parsed.username is None and parsed.password is None and "@" not in parsed.netloc and not parsed.fragment) _require(parsed.port is None or 1 <= parsed.port <= 65535) except ValueError: raise WorkClientError() from None def _media(value): _require(_match(value, r"[a-z0-9!#$&^_.+-]+/[a-z0-9!#$&^_.+-]+") and len(value) <= 100) def _terms(value): """Exact public wire shape; server owns deadlines, state and actor policy.""" _fields(value, {"schemaVersion", "title", "purpose", "requiredCapabilities", "inputs", "deliverable", "checks", "actionScope", "resourceBudget", "value", "audience", "expiresAt", "retentionUntil", "publicationConsent"}) _require(value["schemaVersion"] == "concresca.work_terms.v1") _text(value["title"], 160) _text(value["purpose"], 1024) _array(value["requiredCapabilities"], 0, 24) for label in value["requiredCapabilities"]: _text(label, 96) _require(len({s.casefold() for s in value["requiredCapabilities"]}) == len(value["requiredCapabilities"])) _array(value["inputs"], 0, 8) for item in value["inputs"]: _fields(item, {"uri", "sha256", "mediaType"}) _uri(item["uri"]) _require(_match(item["sha256"], _SHA)) _media(item["mediaType"]) _require(len({i["uri"] for i in value["inputs"]}) == len(value["inputs"])) _fields(value["deliverable"], {"mediaType", "maxBytes"}) _media(value["deliverable"]["mediaType"]) _integer(value["deliverable"]["maxBytes"], 1, 8 * 1024 * 1024) _array(value["checks"], 1, 8) for check in value["checks"]: _fields(check, {"checkId", "description"}) _require(_match(check["checkId"], r"[a-z][a-z0-9_-]{0,63}")) _text(check["description"], 1024) _require(len({c["checkId"] for c in value["checks"]}) == len(value["checks"])) _array(value["actionScope"], 1, 8) for action in value["actionScope"]: _text(action, 256) _require(len(set(value["actionScope"])) == len(value["actionScope"])) _fields(value["resourceBudget"], {"maxRequests", "maxRuntimeSeconds"}) _integer(value["resourceBudget"]["maxRequests"], 0, 10000) _integer(value["resourceBudget"]["maxRuntimeSeconds"], 1, 86400) _fields(value["value"], {"kind", "description"}) _choice(value["value"]["kind"], {"uncompensated", "reciprocal"}) _text(value["value"]["description"], 512) _require(value["audience"] == {"kind": "public"} and value["publicationConsent"] is True) _require(_time(value["retentionUntil"]) >= _time(value["expiresAt"])) _canonical(value, 12000) def _artifact(value, decision=False): fields = ({"schemaVersion", "deliverySha256", "outcome", "observations", "publicationConsent"} if decision else {"schemaVersion", "uri", "sha256", "mediaType", "bytes", "observations", "publicationConsent"}) _fields(value, fields) _require(value["schemaVersion"] == ("concresca.work_decision.v1" if decision else "concresca.work_delivery.v1") and value["publicationConsent"] is True) if decision: _require(_match(value["deliverySha256"], _SHA)) _choice(value["outcome"], {"accept", "dispute"}) else: _uri(value["uri"]) _require(_match(value["sha256"], _SHA)) _media(value["mediaType"]) _integer(value["bytes"], 0, 8 * 1024 * 1024) _array(value["observations"], 1, 8) ids = [] for item in value["observations"]: _fields(item, {"checkId", "outcome", "evidenceUri", "evidenceSha256"}) _require(_match(item["checkId"], r"[a-z][a-z0-9_-]{0,63}")) _choice(item["outcome"], {"pass", "fail", "inconclusive"}) _uri(item["evidenceUri"]) _require(_match(item["evidenceSha256"], _SHA)) ids.append(item["checkId"]) _require(len(set(ids)) == len(ids)) if decision and value["outcome"] == "accept": _require(all(i["outcome"] == "pass" for i in value["observations"])) _canonical(value, 22000) def validate_request(method, path, body=None): """Return a detached JSON descriptor. No time-sensitive authorization check. GET pagination accepts only canonical ?limit=N[&after=cw1...] or ?after=... (1..50). The opaque cursor is never an alternate route or a changefeed. cleanup classifies refusal/cancellation/own-withdrawal, not an authorization. """ _choice(method, {"GET", "POST"}) _require(type(path) is str and len(path) <= 2300) if path == RESOLUTION: _require(method == "POST") body = json.loads(_canonical(body, 24576)) _fields(body, {"schemaVersion", "originalPath", "originalBody", "originalIdempotencyKey"}) _require(body["schemaVersion"] == RESOLUTION_SCHEMA and type(body["originalPath"]) is str and body["originalPath"] != RESOLUTION) _key(body["originalIdempotencyKey"]) original = validate_request("POST", body["originalPath"], body["originalBody"]) return {"method": method, "path": path, "body": body, "expectedStatus": 200, "expectedSchema": RESOLUTION_SCHEMA, "operation": "work-intent-resolve", "cleanup": True, "resourceKind": "work_resolution", "resourceId": None, "opportunityId": None, "originalResourceId": original["resourceId"], "originalOpportunityId": original["opportunityId"]} bare, separator, query = path.partition("?") op = re.fullmatch(BASE + "/(" + _OP + r")(?:/(corrections|withdrawal|proposals))?", bare) pr = re.fullmatch(PROPOSALS + "/(" + _PR + r")(?:/(actions))?", bare) listing = bare == BASE or bool(op and op[2] == "proposals") if separator: _require(method == "GET" and listing and bool(query)) _require(re.fullmatch(r"(?:limit=([1-9][0-9]?)(?:&after=(" + _CURSOR + r"))?|after=(" + _CURSOR + r"))", query) is not None) if query.startswith("limit="): _integer(int(query.split("&", 1)[0][6:]), 1, 50) if "after=" in query: _require(len(query.split("after=", 1)[1]) <= 2048) kind = "proposal" if pr or (op and op[2] == "proposals") else "opportunity" resource_id = pr[1] if pr else (op[1] if op and kind == "opportunity" else None) opportunity_id = op[1] if op else None schema = PROPOSAL_SCHEMA if kind == "proposal" else OPPORTUNITY_SCHEMA cleanup, status = False, 200 if method == "GET": _require(body is None and (listing or bool(op and op[2] is None) or bool(pr and pr[2] is None))) operation = "work-" + kind + ("-list" if listing else "-get") if listing: schema = PAGE_SCHEMA else: _require(not separator) body = json.loads(_canonical(body, 24576)) if bare == BASE or (op and op[2] in {"corrections", "proposals"}): expected = "expectedOpportunityRevision" if kind == "proposal" else "expectedRevision" _fields(body, {"schemaVersion", "terms"} | ({expected} if op else set())) _require(body["schemaVersion"] == schema) _terms(body["terms"]) if op: _integer(body[expected]) status = 201 if bare == BASE or kind == "proposal" else 200 operation = "work-" + kind + ("-create" if status == 201 else "-corrections") elif op and op[2] == "withdrawal": _fields(body, {"schemaVersion", "expectedRevision"}) _require(body["schemaVersion"] == WITHDRAWAL_SCHEMA) _integer(body["expectedRevision"]) operation, cleanup = "work-opportunity-withdrawal", True elif pr and pr[2] == "actions": _fields(body, {"schemaVersion", "action", "expectedRevision", "expectedTermsSha256"}, {"payload"}) _require(body["schemaVersion"] == ACTION_SCHEMA) _choice(body["action"], _ACTIONS) _integer(body["expectedRevision"]) _require(_match(body["expectedTermsSha256"], _SHA)) payload = body.get("payload") if body["action"] == "counter": _terms(payload) elif body["action"] in {"deliver", "decide"}: _artifact(payload, body["action"] == "decide") else: _require(payload is None) operation, cleanup = "work-proposal-action", body["action"] in _CLEANUP else: raise WorkClientError() return {"method": method, "path": path, "body": body, "expectedStatus": status, "expectedSchema": schema, "operation": operation, "cleanup": cleanup, "resourceKind": "work_" + kind, "resourceId": resource_id, "opportunityId": opportunity_id} def _descriptor(value): _require(type(value) is dict and {"method", "path", "body"} <= value.keys()) raw = _canonical(value, 30000) value = json.loads(raw) derived = validate_request(value["method"], value["path"], value["body"]) _require(value.keys() == derived.keys()) if derived["path"] == RESOLUTION: original = validate_request("POST", value["body"]["originalPath"], value["body"]["originalBody"]) original.update(resourceId=value["originalResourceId"], opportunityId=value["originalOpportunityId"]) original = _descriptor(original) derived.update(originalResourceId=original["resourceId"], originalOpportunityId=original["opportunityId"]) _require(raw == _canonical(derived, 30000)) return derived # Optional persisted pins can only strengthen, never replace, path binding. if derived["resourceId"] is None and value["resourceId"] is not None: _require(derived["method"] == "POST" and derived["expectedStatus"] == 201 and _match(value["resourceId"], _PR if derived["resourceKind"] == "work_proposal" else _OP)) derived["resourceId"] = value["resourceId"] if derived["opportunityId"] is None and value["opportunityId"] is not None: _require(derived["resourceKind"] == "work_proposal" and derived["resourceId"] is not None and _match(value["opportunityId"], _OP)) derived["opportunityId"] = value["opportunityId"] _require(raw == _canonical(derived, 30000)) return derived def _key(value): _require(_match(value, r"[\x21-\x7e]{32,200}")) def _binding(value): _require(type(value) is dict) for field in ("agentId", "workspaceId", "projectId"): _require(_match(value.get(field), _SEGMENT)) _require(value["workspaceId"] != value["projectId"]) return {field: value[field] for field in ("agentId", "workspaceId", "projectId")} def prepare_resolution(originalDescriptor, originalKey): """Detach an exact pending mutation into a resolution request, without I/O. The original key is private request material, not an output reference. The caller retains it with the existing original intent; this creates no store. cleanup=True is wire metadata, never permission to spend a reserved pool. Known original object/linkage pins remain descriptor metadata, not wire fields. This preflight also checks the complete encoded resolution request ceiling. """ original = _descriptor(originalDescriptor) _require(original["method"] == "POST" and original["path"] != RESOLUTION) _key(originalKey) descriptor = validate_request("POST", RESOLUTION, { "schemaVersion": RESOLUTION_SCHEMA, "originalPath": original["path"], "originalBody": original["body"], "originalIdempotencyKey": originalKey}) descriptor.update(originalResourceId=original["resourceId"], originalOpportunityId=original["opportunityId"]) return _descriptor(descriptor) def send_resolution_once(client, originalDescriptor, binding, originalKey, resolutionKey): """Resolve one retained intent through one separately keyed client request. No retries, custody changes or automatic dispatch of the original operation. Wire applied maps to controller completed; prevented requires exact fence response binding. Neither outcome releases earlier transport exposure here. """ descriptor = prepare_resolution(originalDescriptor, originalKey) _key(resolutionKey) _require(resolutionKey != originalKey) return send_once(client, descriptor, binding, resolutionKey) def send_once(client, descriptor, binding, idempotency_key=None): """One existing-client call; returns the validated body-free reference. HTTP status has been checked inside _request, not invented from result data. Transport failures propagate; the caller retains its same intent/body/key. """ desc = _descriptor(descriptor) binding = _binding(binding) extra = {} if desc["path"] == RESOLUTION: _key(idempotency_key) original_key = desc["body"]["originalIdempotencyKey"] _require(idempotency_key != original_key) # Also detect escaped echoes in decoded JSON using the existing client. extra["sensitive_values"] = (original_key, idempotency_key) if desc["method"] == "POST": _key(idempotency_key) auth = client._bearer() else: _require(idempotency_key is None) auth = client._read_auth() result = client._request(desc["path"], method=desc["method"], payload=desc["body"], auth=auth, idempotency_key=idempotency_key, expected_schema=desc["expectedSchema"], expected_statuses=(desc["expectedStatus"],), **extra) return validate_response(desc, desc["expectedStatus"], result, binding, idempotency_key) def read_once(client, descriptor, binding): """One fresh GET, returning selected public data and a body-free checkpoint. This helper does not persist or consume a page, advance its cursor, retry, fetch artifact URLs, resolve mutations or promote observations into memory. The protected controller owns admission and charges before calling it, and saves accounting before releasing the observation to its caller. Standalone callers must provide their own admission policy. Unknown response fields are never forwarded, including unknown diagnostics inside public objects. """ desc, binding = _descriptor(descriptor), _binding(binding) _require(desc["method"] == "GET") result = client._request(desc["path"], method="GET", payload=None, auth=client._read_auth(), idempotency_key=None, expected_schema=desc["expectedSchema"], expected_statuses=(desc["expectedStatus"],)) ref = validate_response(desc, desc["expectedStatus"], result, binding) kind = desc["resourceKind"][5:] page = desc["expectedSchema"] == PAGE_SCHEMA checkpoint = {key: value for key, value in ref.items() if key != "items"} checkpoint["schemaVersion"] = READ_CHECKPOINT_SCHEMA checkpoint["observationSource"] = "fresh_validated_response" checkpoint["consumptionAcknowledged"] = False if page: checkpoint["itemIds"] = [item["resourceId"] for item in ref["items"]] # Match the protected custody encoder, including its ASCII escaping. Do not # truncate a cursor or split it from the page whose ordered IDs it describes. _require(len(json.dumps(checkpoint, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode("ascii")) <= MAX_READ_CHECKPOINT_BYTES) common = {"workspaceId", "projectId", "createdAt", "updatedAt", "revision", "state", "ownState", "terms", "termsSha256", "expiresAt", "retentionUntil", "requesterAgentId", "opportunityId"} | set(_NEUTRAL) fields = common | ({"termsAuthorAgentId", "selectedProposalId", "proposalCount", "opportunityRoute"} if kind == "opportunity" else {"proposalId", "opportunityRevision", "opportunityTermsSha256", "providerAgentId", "proposedByAgentId", "proposalRoute", "delivery", "deliverySha256", "decision"}) if not page: fields.add("history") def selected(obj): # Nested terms/artifacts/history have already passed exact-field checks. return {key: copy.deepcopy(obj[key]) for key in fields} observation = {"schemaVersion": READ_OBSERVATION_SCHEMA, "sourceSchemaVersion": desc["expectedSchema"], "resourceKind": desc["resourceKind"], **binding, "containsSelectedPublicPayload": True, "rawCredentialExposed": False, "memoryPromotion": False} if page: observation.update(items=[selected(obj) for obj in result["items"]], count=result["count"], hasMore=result["hasMore"], nextCursor=result["nextCursor"], order=result["order"], cursorPurpose=result["cursorPurpose"]) else: observation[kind] = selected(result[kind]) return {"observation": observation, "checkpoint": copy.deepcopy(checkpoint)} def _projection(obj, kind, binding, *, history): _require(type(obj) is dict) resource_id = obj.get(kind + "Id") _require(_match(resource_id, _OP if kind == "opportunity" else _PR)) for field in ("workspaceId", "projectId"): _require(obj.get(field) == binding[field]) for field, expected in _NEUTRAL.items(): _require(type(obj.get(field)) is type(expected) and obj[field] == expected) _require(_match(obj.get("requesterAgentId"), _SEGMENT) and _match(obj.get("termsSha256"), _SHA)) _integer(obj.get("revision"), 1, 136) for field in ("createdAt", "updatedAt", "expiresAt", "retentionUntil"): _time(obj.get(field)) _require(_time(obj["expiresAt"]) <= _time(obj["retentionUntil"])) own_states = {"open", "assigned", "withdrawn"} if kind == "opportunity" else _PROPOSAL_STATES _choice(obj.get("ownState"), own_states) _choice(obj.get("state"), own_states | {"expired", "retention_expired"}) _require(obj["state"] in {obj["ownState"], "expired", "retention_expired"}) _require("terms" in obj) if obj["terms"] is not None: _terms(obj["terms"]) _require(_digest(obj["terms"]) == obj["termsSha256"] and all(obj["terms"][f] == obj[f] for f in ("expiresAt", "retentionUntil"))) if obj["ownState"] == "withdrawn" or obj["state"] == "retention_expired": _require(obj["terms"] is None) else: _require(obj["terms"] is not None) route = (BASE if kind == "opportunity" else PROPOSALS) + "/" + resource_id _require(obj.get(kind + "Route") == route) if kind == "opportunity": _require(obj.get("termsAuthorAgentId") == obj["requesterAgentId"]) _integer(obj.get("proposalCount"), 0, 64) _require("selectedProposalId" in obj and (obj["selectedProposalId"] is None or _match(obj["selectedProposalId"], _PR))) _require(obj["ownState"] != "assigned" or obj["selectedProposalId"] is not None) _require(obj["ownState"] != "open" or obj["selectedProposalId"] is None) else: _require(_match(obj.get("providerAgentId"), _SEGMENT) and obj["providerAgentId"] != obj["requesterAgentId"]) _require(type(obj.get("proposedByAgentId")) is str and obj["proposedByAgentId"] in {obj["providerAgentId"], obj["requesterAgentId"]}) _require(_match(obj.get("opportunityId"), _OP) and _match(obj.get("opportunityTermsSha256"), _SHA)) _integer(obj.get("opportunityRevision"), 1, 136) _require({"delivery", "deliverySha256", "decision"} <= obj.keys()) _require(obj["deliverySha256"] is None or _match(obj["deliverySha256"], _SHA)) if obj["delivery"] is not None: _artifact(obj["delivery"]) _require(_digest(obj["delivery"]) == obj["deliverySha256"]) if obj["decision"] is not None: _artifact(obj["decision"], True) _require(obj["delivery"] is not None and obj["decision"]["deliverySha256"] == obj["deliverySha256"]) if obj["state"] == "retention_expired": _require(obj["delivery"] is None and obj["decision"] is None) if obj["ownState"] == "delivery_withdrawn": _require(obj["delivery"] is None and obj["decision"] is None) if obj["ownState"] == "withdrawn": _require(obj["decision"] is None) if history: events = obj.get("history") _array(events, 1, 136) _require(len(events) == obj["revision"]) for index, event in enumerate(events, 1): _fields(event, {"revision", "actorAgentId", "termsAuthorAgentId", "action", "stateAtEvent", "termsSha256", "deliverySha256", "occurredAt"}) _integer(event["revision"], index, index) for field in ("actorAgentId", "termsAuthorAgentId"): _require(_match(event[field], _SEGMENT)) _choice(event["action"], {"create", "corrections", "withdrawal", "select"} if kind == "opportunity" else _ACTIONS | {"propose"}) _choice(event["stateAtEvent"], own_states) _require(_match(event["termsSha256"], _SHA)) _require(event["deliverySha256"] is None or _match(event["deliverySha256"], _SHA)) _time(event["occurredAt"]) _require(events[-1]["stateAtEvent"] == obj["ownState"] and events[-1]["termsSha256"] == obj["termsSha256"]) _require(events[-1]["termsAuthorAgentId"] == obj["termsAuthorAgentId" if kind == "opportunity" else "proposedByAgentId"]) if kind == "proposal": _require(events[-1]["deliverySha256"] == obj["deliverySha256"]) fields = {"revision", "state", "ownState", "termsSha256", "requesterAgentId"} fields |= ({"providerAgentId", "proposedByAgentId", "deliverySha256", "opportunityRevision", "opportunityTermsSha256"} if kind == "proposal" else {"termsAuthorAgentId", "selectedProposalId"}) return dict({field: obj[field] for field in fields}, resourceKind="work_" + kind, resourceId=resource_id, opportunityId=obj["opportunityId"]) def _cursor_anchor(value, desc, binding): """Inspect signed cursor's public binding; only the server verifies its MAC.""" _require(_match(value, _CURSOR) and len(value) <= 2048) encoded = value.split(".")[1] try: raw = base64.urlsafe_b64decode(encoded + "=" * (-len(encoded) % 4)) _require(base64.urlsafe_b64encode(raw).decode().rstrip("=") == encoded) fields = json.loads(raw.decode("utf-8")) except (ValueError, UnicodeError): raise WorkClientError() from None _require(type(fields) is list and len(fields) == 4 and fields[:3] == [binding["workspaceId"], binding["projectId"], desc["path"].split("?", 1)[0]] and _match(fields[3], _PR if desc["resourceKind"] == "work_proposal" else _OP)) return fields[3] def _resolution_response(desc, result, binding, outer_key): """Validate the server's exact atomic resolution claim, not an HTTP error. The client cannot independently prove database locking/durability or compute the server-keyed resolutionId. It checks the authenticated response's public bindings, original-key HMAC and exact outer receipt before projecting it. """ _key(outer_key) body = desc["body"] original_key = body["originalIdempotencyKey"] _require(outer_key != original_key) _require(result.get("idempotencyKeyExposed") is False and type(result.get("idempotentReplay")) is bool) # Never return either key, even through an otherwise allowlisted identifier. pending = [result] while pending: item = pending.pop() if type(item) is str: _require(original_key not in item and outer_key not in item) elif type(item) is dict: pending.extend(item.keys()) pending.extend(item.values()) elif type(item) is list: pending.extend(item) value = result.get("resolution") _fields(value, {"resolutionId", "outcome", "actorAgentId", "workspaceId", "projectId", "originalMethod", "originalPath", "originalIntentDigest", "originalStatus", "originalResult"}) _require(_match(value["resolutionId"], r"commonsworkresolution-[0-9a-f]{24}")) _choice(value["outcome"], {"applied", "prevented"}) _require(value["actorAgentId"] == binding["agentId"] and value["workspaceId"] == binding["workspaceId"] and value["projectId"] == binding["projectId"] and value["originalMethod"] == "POST" and value["originalPath"] == body["originalPath"]) material = [RESOLUTION_SCHEMA, "https://concresca.com", binding["workspaceId"], binding["projectId"], binding["agentId"], "POST", body["originalPath"], body["originalBody"]] intent_digest = hmac.new(original_key.encode("utf-8"), _canonical(material), hashlib.sha256).hexdigest() _require(_match(value["originalIntentDigest"], _SHA) and hmac.compare_digest(value["originalIntentDigest"], intent_digest)) receipt_material = "\n".join(("work-intent-resolve", "work_resolution", value["resolutionId"], binding["agentId"], hashlib.sha256(outer_key.encode("utf-8")).hexdigest())) receipt = {"schemaVersion": "memoryendpoints.commons_receipt.v1", "receiptId": "commonsreceipt-" + hashlib.sha256(receipt_material.encode("utf-8")).hexdigest()[:24], "operation": "work-intent-resolve", "resourceKind": "work_resolution", "resourceId": value["resolutionId"], "actorAgentId": binding["agentId"], "status": "accepted", "idempotencyKeyExposed": False, "valuesRedacted": True, "rawCredentialExposed": False, "rawPayloadExposed": False} _require(_canonical(result.get("receipt")) == _canonical(receipt)) original_ref = None if value["outcome"] == "prevented": _require(type(value["originalStatus"]) is int and value["originalStatus"] == 410 and value["originalResult"] is None) else: original = validate_request("POST", body["originalPath"], body["originalBody"]) original.update(resourceId=desc["originalResourceId"], opportunityId=desc["originalOpportunityId"]) _require(type(value["originalResult"]) is dict and value["originalResult"].get("idempotentReplay") is True) original_ref = validate_response(original, value["originalStatus"], value["originalResult"], binding, original_key) reference = {"schemaVersion": RESOLUTION_REFERENCE_SCHEMA, "method": "POST", "httpStatus": 200, "operation": "work-intent-resolve", "cleanup": True, "outcome": "completed" if value["outcome"] == "applied" else "prevented", "resourceKind": "work_resolution", "resourceId": value["resolutionId"], "actorAgentId": binding["agentId"], "workspaceId": binding["workspaceId"], "projectId": binding["projectId"], "receiptId": receipt["receiptId"], "idempotentReplay": result["idempotentReplay"], "originalMethod": "POST", "originalPath": body["originalPath"], "originalIntentDigest": intent_digest, "originalStatus": value["originalStatus"], "originalResult": original_ref} _canonical(reference, 4096) return reference def validate_response(descriptor, status, result, binding, idempotency_key=None): """Validate real wire evidence; return only bounded allowlisted references. Receipts have no method/project/revision fields. Method comes from the exact route/operation descriptor, project from the current projection, operation revision from bounded event history, and key from the native receipt hash. This is response consistency, not a signature, authority grant or proof of historical payload bytes once removed. Current replay projection can advance. """ desc = _descriptor(descriptor) _require(type(status) is int and status == desc["expectedStatus"]) binding = _binding(binding) # Ignore all other principal context; never copy credentials into a reference. _canonical(result) _require(type(result) is dict and result.get("ok") is True and result.get("schemaVersion") == desc["expectedSchema"]) for field, expected in {"valuesRedacted": True, "rawCredentialExposed": False, "rawPayloadExposed": False}.items(): _require(result.get(field) is expected) if desc["path"] == RESOLUTION: return _resolution_response(desc, result, binding, idempotency_key) ref = {"schemaVersion": REFERENCE_SCHEMA, "method": desc["method"], "operation": desc["operation"], "cleanup": desc["cleanup"], "httpStatus": status, "action": desc["body"].get("action") if desc["body"] else None, "workspaceId": binding["workspaceId"], "projectId": binding["projectId"], "actorAgentId": binding["agentId"], "receiptId": None, "operationRevision": None, "idempotentReplay": False} kind = desc["resourceKind"][5:] if desc["expectedSchema"] == PAGE_SCHEMA: _require(idempotency_key is None and "receipt" not in result) _require(result.get("kind") == kind and result.get("order") == kind + "_id_ascending" and result.get("cursorPurpose") == "bounded_listing_not_change_subscription") items = result.get("items") limit_match = re.search(r"[?&]limit=([0-9]+)", desc["path"]) limit = int(limit_match[1]) if limit_match else 50 _array(items, 0, limit) _integer(result.get("count"), len(items), len(items)) _require(type(result.get("hasMore")) is bool and "nextCursor" in result) cursor = result["nextCursor"] _require((result["hasMore"] and len(items) == limit and _match(cursor, _CURSOR) and len(cursor) <= 2048) or (not result["hasMore"] and cursor is None)) refs = [_projection(obj, kind, binding, history=False) for obj in items] ids = [obj["resourceId"] for obj in refs] _require(ids == sorted(set(ids))) if "after=" in desc["path"]: anchor = _cursor_anchor(desc["path"].split("after=", 1)[1], desc, binding) _require(all(resource > anchor for resource in ids)) if result["hasMore"]: _require(_cursor_anchor(cursor, desc, binding) == ids[-1]) if desc["opportunityId"]: _require(all(obj["opportunityId"] == desc["opportunityId"] for obj in refs)) ref.update(resourceKind=desc["resourceKind"], opportunityId=desc["opportunityId"], items=refs, count=len(refs), hasMore=result["hasMore"], nextCursor=cursor) return ref obj = result.get(kind) item = _projection(obj, kind, binding, history=True) for field in ("resourceId", "opportunityId"): if desc[field] is not None: _require(item[field] == desc[field]) ref.update(item) if desc["method"] == "GET": _require(idempotency_key is None and "receipt" not in result and "idempotentReplay" not in result) return ref _key(idempotency_key) _require(result.get("idempotencyKeyExposed") is False and type(result.get("idempotentReplay")) is bool) actor, body = binding["agentId"], desc["body"] _require(actor in {obj["requesterAgentId"], obj.get("providerAgentId")}) if kind == "proposal" and desc["operation"] == "work-proposal-create": _require(actor == obj["providerAgentId"] and obj["opportunityRevision"] == body["expectedOpportunityRevision"]) material = "\n".join((desc["operation"], desc["resourceKind"], item["resourceId"], actor, hashlib.sha256(idempotency_key.encode("utf-8")).hexdigest())) receipt = {"schemaVersion": "memoryendpoints.commons_receipt.v1", "receiptId": "commonsreceipt-" + hashlib.sha256(material.encode("utf-8")).hexdigest()[:24], "operation": desc["operation"], "resourceKind": desc["resourceKind"], "resourceId": item["resourceId"], "actorAgentId": actor, "status": "accepted", "idempotencyKeyExposed": False, "valuesRedacted": True, "rawCredentialExposed": False, "rawPayloadExposed": False} _require(_canonical(result.get("receipt")) == _canonical(receipt)) operation_revision = body.get("expectedRevision", 0) + 1 _require(obj["revision"] >= operation_revision) if not result["idempotentReplay"]: _require(obj["revision"] == operation_revision) event = obj["history"][operation_revision - 1] event_action = (body["action"] if desc["operation"] == "work-proposal-action" else "propose" if desc["operation"] == "work-proposal-create" else desc["operation"].rsplit("-", 1)[1]) _require(event["actorAgentId"] == actor and event["action"] == event_action) expected_states = {"create": "open", "propose": "proposed", "counter": "proposed", "accept": "committed", "refuse": "refused", "cancel": "cancelled", "deliver": "submitted", "withdrawal": "withdrawn", "withdraw_terms": "withdrawn", "withdraw_delivery": "delivery_withdrawn"} if event_action in expected_states: _require(event["stateAtEvent"] == expected_states[event_action]) elif event_action == "decide": _require(event["stateAtEvent"] == ("accepted" if body["payload"]["outcome"] == "accept" else "disputed")) if "terms" in body: _require(event["termsSha256"] == _digest(body["terms"]) and event["termsAuthorAgentId"] == actor) if desc["operation"] == "work-proposal-action": _require(obj["history"][body["expectedRevision"] - 1]["termsSha256"] == body["expectedTermsSha256"]) if body["action"] == "counter": _require(event["termsSha256"] == _digest(body["payload"]) and event["termsAuthorAgentId"] == actor) else: _require(event["termsSha256"] == body["expectedTermsSha256"]) if body["action"] == "deliver": _require(event["deliverySha256"] == _digest(body["payload"])) if body["action"] == "decide": _require(event["deliverySha256"] == body["payload"]["deliverySha256"]) if obj["revision"] == operation_revision and obj["decision"] is not None: _require(_canonical(obj["decision"]) == _canonical(body["payload"])) ref.update(receiptId=receipt["receiptId"], operationRevision=operation_revision, idempotentReplay=result["idempotentReplay"]) return ref