commerce-agents 把购物流程做成可审计 Agent 蓝图

  17 
  18 RecordT = TypeVar("RecordT")
  19 
  20 
  21 def remember(records: dict[str, RecordT], key: str, value: RecordT) -> None:
  22     records.pop(key, None)
  23     records[key] = value
  24     while len(records) > PROVENANCE_CAP:
  25         del records[next(iter(records))]
  26 
  27 
  28 class MemoryCategory(StrEnum):
  29     PREFERENCE = "preference"
  30     CONSTRAINT = "constraint"
  31     CONTEXT = "context"
  32 
  33 
  34 class MemoryFact(BaseModel):
  35     """One stored fact. The schema bounds the shape; ``validate_fact`` decides what a
  36     value may be about. Constraints are injected on every turn."""
  37 
  38     key: str = Field(max_length=64)
  39     value: str = Field(max_length=200)
  40     category: MemoryCategory = MemoryCategory.PREFERENCE
  41     updated_at: datetime | None = None
  42     # Recalled facts carry the session that wrote them, so the model reads them as a
  43     # past turn's claim and a poisoned session's writes stay traceable. The value is the
  44     # session's ``session_tag``, not the id: a fact is read back by other sessions of the
  45     # subject, and on the example hosts the id is also the request credential.
  46     source_session_id: str | None = Field(default=None, max_length=80)
  47 
  48 

 270     def feed(self, raw: Any) -> StreamedTool | None:
 271         """Record one raw stream event; returns the tool it touched, if any."""
 272         raw_type = getattr(raw, "type", "")
 273         if raw_type in ("message_start", "message_delta"):
 274             # The stream's counts are cumulative, so the latest value of each replaces ours.
 275             counts = getattr(raw.message if raw_type == "message_start" else raw, "usage", None)
 276             for key in usage_totals():
 277                 if isinstance(value := getattr(counts, key, None), int):
 278                     setattr(self.usage, key, value)
 279             return None
 280         index = getattr(raw, "index", None)
 281         if not isinstance(index, int):
 282             return None
 283         if raw_type == "content_block_start":
 284             block = raw.content_block
 285             entry = dict(block.model_dump(exclude_none=True, exclude={"citations"}))
 286             kind = entry.get("type")
 287             # The streamed fields start empty here and grow from the deltas.
 288             if kind == "text":
 289                 entry["text"] = ""
 290             elif kind == "thinking":
 291                 entry["thinking"] = ""
 292             elif kind in ("tool_use", "server_tool_use"):
 293                 entry["input"] = {}
 294                 self.tools[index] = StreamedTool(
 295                     name=block.name, id=block.id, server=kind == "server_tool_use"
 296                 )
 297             self.blocks.append(entry)
 298             return self.tools.get(index)
 299         if raw_type == "content_block_delta":
 300             delta = raw.delta
 301             delta_type = getattr(delta, "type", "")
 302             if (tool := self.tools.get(index)) is not None:
 303                 if delta_type == "input_json_delta":
 304                     tool.buffer += delta.partial_json
 305                 return tool
 306             entry = self.blocks[index] if index < len(self.blocks) else {}
 307             if delta_type == "text_delta":
 308                 entry["text"] = entry.get("text", "") + delta.text
 309             elif delta_type == "thinking_delta":
 310                 entry["thinking"] = entry.get("thinking", "") + delta.thinking
 311             elif delta_type == "signature_delta":
 312                 entry["signature"] = delta.signature
 313             return None
 314         if raw_type == "content_block_stop" and (tool := self.tools.get(index)) is not None:
 315             tool.closed = True
 316             if tool.server and (args := tool.parsed()) is not None:
 317                 self.blocks[index]["input"] = args
 318             return tool
 319         return None

 214     async def execute(self, name: str, tool_input: dict[str, Any] | None) -> ToolOutcome:
 215         try:
 216             return await self.dispatch(name, dict(tool_input or {}))
 217         except InvalidArguments as invalid:
 218             return ToolOutcome.error(invalid_arguments_text(name, invalid.invalid))
 219         except Exception as error:  # a tool failure must not end the turn
 220             if (outcome := self.domain_error(error)) is not None:
 221                 return outcome
 222             logger.warning("tool %s failed and is reported as unavailable", name, exc_info=True)
 223             return ToolOutcome.error(self.unavailable_text.format(name=name))
 224 
 225     async def dispatch(self, name: str, tool_input: dict[str, Any]) -> ToolOutcome:
 226         """:meth:`execute` without the failure ladder: what the tool raised propagates,
 227         so a host prefetching a read can tell a failed tool from a result the tool wrote
 228         (a not-found line, a held call). The ``status`` line is dropped here, before any
 229         argument is validated or any handler runs; a tool the deployment's config leaves
 230         out of the tool list is unknown here too, whichever path calls it."""
 231         tool_input, _status = self.split_status(name, tool_input)
 232         if name in self._absent:
 233             return ToolOutcome.error(self.absent_text.format(name=name))
 234         if name == LOAD_SKILL:
 235             return self._load_skill(tool_input)
 236         if (spec := self.components.get(name) or self._extensions.get(name)) is not None:
 237             return await self._present(spec, tool_input)
 238         if (delegate := self._delegates.get(name)) is not None:
 239             return await self._run_delegate(delegate, tool_input)
 240         handler = self._handlers.get(name)
 241         if handler is None:
 242             return ToolOutcome.error(f"Unknown tool: {name}")
 243         return await handler(tool_input)

 102 async def gated_add_to_cart(
 103     *,
 104     backend: StorefrontBackend,
 105     config: ShoppingAgentConfig,
 106     session: ShoppingSessionContext,
 107     state: ShoppingSessionState,
 108     product_id: str,
 109     quantity: int,
 110 ) -> ToolOutcome:
 111     if held := check_provenance(state, product_id) or check_options(state, product_id):
 112         return held
 113     requested = max(1, quantity)
 114     max_quantity = config.max_quantity_per_item
 115     async with _cart_lock(session):
 116         current = await backend.get_cart(session)
 117         existing = next((i for i in current.items if i.product_id == product_id), None)
 118         if existing is None and len(current.items) >= config.max_cart_lines:
 119             return ToolOutcome.error("The cart is full.")
 120         allowed = min(requested, max(0, max_quantity - (existing.quantity if existing else 0)))
 121         if allowed <= 0:
 122             return ToolOutcome.error(
 123                 f"This item is already at the per-item limit of {max_quantity}."
 124             )
 125         cart = await backend.add_to_cart(session, product_id, allowed)
 126     # The confirmation names the id only: titles are catalog text and stay inside fences.
 127     capped = f" (capped at the per-item limit of {max_quantity})" if allowed < requested else ""
 128     return _written(
 129         f"Added {product_id} x{allowed}{capped}. Cart now has {cart_summary(cart)}.", cart
 130     )

 192 def check_apply_change(
 193     state: MerchantSessionState, config: MerchantAgentConfig, change_id: str
 194 ) -> ToolOutcome | None:
 195     """Provenance, then the guardrails under the current config, then the host's
 196     approval mark when the deployment requires one."""
 197     known = state.seen_changes.get(change_id)
 198     if known is None:
 199         return ToolOutcome.held(
 200             PROVENANCE_GATE,
 201             f"change_id {change_id} was not staged or listed in this session. Stage "
 202             "the change (or call get_pending_changes) first, preview it, and apply it "
 203             "only after the operator approves it.",
 204         )
 205     if violations := check_guardrails(known.kind, known.items, config):
 206         return ToolOutcome.held(GUARDRAIL_GATE, apply_guardrail_message(violations))
 207     if config.require_host_approval and change_id not in state.approved_change_ids:
 208         return ToolOutcome.held(
 209             APPROVAL_GATE,
 210             f"change {change_id} has not been approved through "
 211             f"{config.approval_surface}. Tell the operator it is staged and "
 212             f"waiting for their approval on {config.approval_surface} — "
 213             "approving it there is what applies it.",
 214         )
 215     return None

 112 async def test_update_and_remove_require_provenance_or_cart_membership(executor, backend):
 113     update = await executor.execute("update_cart_item", {"product_id": "p-100", "quantity": 2})
 114     assert update.blocked == "provenance"
 115     # The backend never sees the id; an upsert-style update would otherwise create the line.
 116     assert backend.cart_items == {}
 117     remove = await executor.execute("remove_from_cart", {"product_id": "p-100"})
 118     assert remove.blocked == "provenance"
 119 
 120     await executor.execute("search_products", {"query": "tent"})
 121     await executor.execute("add_to_cart", {"product_id": "p-100", "quantity": 1})
 122     update = await executor.execute("update_cart_item", {"product_id": "p-100", "quantity": 3})
 123     assert not update.is_error and update.blocked is None
 124     assert backend.cart_items["p-100"].quantity == 3
 125 
 126 
 127 async def test_cart_membership_alone_grants_update_and_remove(
 128     backend, config, skills, session, state
 129 ):
 130 
 131     backend.cart_items["p-200"] = CartItem(
 132         product_id="p-200", title="Two-Burner Camp Stove", price=64.5, quantity=2
 133     )
 134     executor = ShoppingToolExecutor(
 135         backend=backend,
 136         config=config,
 137         skills=skills,
 138         session=session,
 139         state=state,
 140         memory=build_memory(config, InMemoryMemoryStore()),
 141     )
 142     update = await executor.execute("update_cart_item", {"product_id": "p-200", "quantity": 4})
 143     assert not update.is_error and update.blocked is None
 144     remove = await executor.execute("remove_from_cart", {"product_id": "p-200"})
 145     assert not remove.is_error and remove.blocked is None
 146     assert backend.cart_items == {}