feat(decisions): add an OpenAI Decisions backend for the decision slots (#7127)

* feat(decisions): add an OpenAI Decisions backend for the decision slots

Issue #7005 asked for a second decision provider behind the interfaces the
TypeSafe (Jev) integration already uses, not a new decision system.
`deerflow/openai_decisions` speaks `POST /v1/decisions`: a `noul` question
becomes a `predicate` whose true/false rubric folds into `instructions` (the
endpoint has no criteria field, and dropping it would move the decision
boundary while the adapter kept reporting the same criteria in its policy
identity), answers arrive as an array rather than a mapping, and a refusal is an
answer `type` rather than an HTTP failure -- so it lands in the existing
per-question error channel and each consumer keeps its own failure direction.

Transport is reused rather than copied. `provider_label` and the connection's
`endpoint_path` are the only two seams a dialect needs, so retry and backoff,
the deadline budget, the bounded never-decoded body read and the error taxonomy
keep one owner; `resolve_decisions_connection` pairs the two so a Decisions
connection cannot be built pointing at System One. The three adapters override
their connection resolution, their client, and the value the endpoint judges
(`input` text instead of a `state` object), and inherit thresholds, rubrics,
caches and audit policy unchanged. Wiring stays a `use:` class path, so no
config schema is added.

* fix(decisions): isolate endpoint identities and correct diagnostics

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
李刚 / Bryan Li 2026-10-20 11:50:46 +08:00 committed by GitHub
parent f2c2490d26
commit 1fd266e0c0
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
16 changed files with 1217 additions and 63 deletions

View File

@ -2906,6 +2906,12 @@ snapshots and each doubling. These snapshots remain admitted beyond `queue_max_d
and stay in memory until processing. Check extraction latency when warnings appear;
process exit can still lose pending work.
Optional memory pre-screening and signal classification support TypeSafe (Jev)
and OpenAI Decisions. With `combine: auto`, their different endpoint paths keep
requests separate even behind the same gateway URL, model, and credentials.
Compatible requests to the same endpoint can still share one call. See the
[decision provider guide](backend/docs/GUARDRAILS.md).
The opt-in [DeerMem scope-isolation benchmark](backend/scripts/benchmark/deermem_scope_isolation/README.md)
checks semantic safety across facts and summaries, and fact routing across users
and agents. Failed extraction attempts are retryable execution errors, not safety passes.

View File

@ -544,6 +544,23 @@ make dev # then ask the agent to delete the disposable file with bash
Check the denial message the agent receives and the `middleware:guardrail` journal event. Do not run this against real project data or a production key.
**The same gate over another endpoint: OpenAI Decisions.** `deerflow.guardrails.openai_decisions:OpenAIDecisionsGuardrailProvider` asks one `predicate` question about the call at `POST {base_url}/v1/decisions` and denies at the same `threshold`. It is the same *gate*, not a re-implementation: the strict-JSON local preflight, `max_state_chars` as a character count, the cache, the fail-closed rule and the whole transport (auth, retry, deadline, bounded body) are inherited from the Option 4 provider and `deerflow.typesafe`, so a change to the security boundary lands once. What differs is the envelope:
| | TypeSafe (Jev) | OpenAI Decisions |
|---|---|---|
| Endpoint | `POST /v1/systemone` | `POST /v1/decisions` |
| Judged value | `state.tool_call` (name + argument JSON) | `input`: the same canonical argument JSON as text |
| Question | `noul` with a `criteria` map | `predicate`; the endpoint has no criteria field, so the rubric rides inside `instructions` |
| Refusal | not a response shape | `{"type": "refusal"}` — a per-question outcome, so the gate still denies for want of a verdict |
| `model` | defaults to `jev-latest` | **required**: no default model is shipped, so a deployment cannot send tool text to a model nobody named |
`use:` is the only wiring — `guardrails.provider.config` keeps accepting the same fields, and there is no new config schema. The memory pre-screen and signal classification have siblings in the same shape (`deerflow.agents.memory.prescreen.openai_decisions:OpenAIDecisionsMemoryPrescreen`, `deerflow.agents.memory.signals.openai_decisions:OpenAIDecisionsSignalClassifier`); their thresholds, modes, cache buckets and `request_failed` reporting are the coordinator's and unchanged, and a `sharing_key` difference keeps a Decisions side from ever being combined into a Jev request. Offline suite: `tests/test_openai_decisions.py`.
The endpoint path participates in the sharing key and public policy identity, so
`combine: auto` keeps Jev and Decisions requests separate even when their gateway
base URL, model, credentials, and transport match. Matching endpoints can still
combine. Missing credential and model errors name the configured service.
## Implementing a Provider
### Required Interface

View File

@ -0,0 +1,58 @@
"""OpenAI Decisions memory pre-screen: the same cost gate over ``/v1/decisions``.
The gate's meaning is unchanged and lives in the base class: it answers "is this batch worth
paying an extraction call for?", it gates no execution and no write, and every failure direction
still extracts. Only the endpoint differs, so this subclass overrides the three things that
cannot be shared — which service the connection resolves against, which client carries it, and
what the provider is handed to judge — and inherits the rubric, the threshold, the cache and the
audit policy.
Because the answers keep the shared ``NoulAnswer`` / ``AnswerSet`` shape, a deployment can move a
side from Jev to Decisions by editing its ``use:`` path alone, and the coordinator's combining,
cache-bucket and ``request_failed`` behaviour is untouched.
Configuration::
memory:
backend_config:
prescreen:
mode: shadow
use: deerflow.agents.memory.prescreen.openai_decisions:OpenAIDecisionsMemoryPrescreen
config:
api_key_env: OPENAI_API_KEY
model: <the decision model this deployment names>
"""
from __future__ import annotations
from deerflow.agents.memory.prescreen.contract import CONFIGURATION_SOURCE
from deerflow.agents.memory.prescreen.typesafe import TypeSafeMemoryPrescreen
from deerflow.openai_decisions.client import OpenAIDecisionsClient, resolve_decisions_connection
from deerflow.typesafe.client import TransportFactory, TypeSafeClient
from deerflow.typesafe.connection import TypeSafeConnection
class OpenAIDecisionsMemoryPrescreen(TypeSafeMemoryPrescreen):
"""Skip the extraction call on a Decisions ``predicate`` probability below ``skip_threshold``."""
name = "openai_decisions"
policy_id = "deerflow.memory.prescreen.openai_decisions"
policy_version = "1.0.0"
def _resolve_connection(self, settings: dict[str, object]) -> TypeSafeConnection:
return resolve_decisions_connection(settings=settings, configuration_source=CONFIGURATION_SOURCE)
def _build_client(self, connection: TypeSafeConnection, *, transport_factory: TransportFactory | None) -> TypeSafeClient:
return OpenAIDecisionsClient(connection, transport_factory=transport_factory)
def _judge_state(self, batch_text: str) -> str:
"""The batch text itself: this endpoint's ``input`` is text, not a state object.
What is judged stays what the Jev side judges — the same
``format_conversation_for_update`` output, bounded by the same ``max_state_chars``
character count — only the envelope around it differs.
"""
return batch_text
__all__ = ["OpenAIDecisionsMemoryPrescreen"]

View File

@ -83,8 +83,8 @@ class TypeSafeMemoryPrescreen:
if mode not in MODES:
raise ValueError(f"memory.prescreen.mode must be one of {list(MODES)}, got {mode!r}")
self._mode = mode
connection: TypeSafeConnection = resolve_connection(
settings={
self._connection = self._resolve_connection(
{
"api_key": api_key,
"api_key_env": api_key_env,
"base_url": base_url,
@ -93,12 +93,9 @@ class TypeSafeMemoryPrescreen:
"deadline_seconds": deadline_seconds,
"max_attempts": max_attempts,
"retry_backoff": retry_backoff,
},
defaults=typesafe_defaults(),
configuration_source=CONFIGURATION_SOURCE,
}
)
self._connection = connection
self._client = TypeSafeClient(connection, transport_factory=transport_factory)
self._client = self._build_client(self._connection, transport_factory=transport_factory)
self.skip_threshold = finite_float("skip_threshold", skip_threshold, minimum=0.0, maximum=1.0)
self._instructions = defaulted_text("instructions", instructions, _DEFAULT_INSTRUCTIONS)
self._criteria_true = defaulted_text("criteria.true", criteria_entry(criteria, True), _DEFAULT_CRITERIA_TRUE)
@ -115,6 +112,19 @@ class TypeSafeMemoryPrescreen:
}
self._cache = AnswerCache(size=self.cache_size, ttl_seconds=self.cache_ttl_seconds)
# --- dialect hooks ------------------------------------------------------
def _resolve_connection(self, settings: dict[str, object]) -> TypeSafeConnection:
"""Resolve this provider's effective connection: consumer config, then its own defaults."""
return resolve_connection(settings=settings, defaults=typesafe_defaults(), configuration_source=CONFIGURATION_SOURCE)
def _build_client(self, connection: TypeSafeConnection, *, transport_factory: TransportFactory | None) -> TypeSafeClient:
return TypeSafeClient(connection, transport_factory=transport_factory)
def _judge_state(self, batch_text: str) -> object:
"""What the provider judges. A dialect whose endpoint takes plain input text overrides it."""
return conversation_tail_state(batch_text)
# --- side interface (shared with the coordinator) ---------------------
def questions(self) -> Mapping[str, Question]:
@ -123,7 +133,7 @@ class TypeSafeMemoryPrescreen:
def ask(self, batch_text: str, questions: Mapping[str, Question]):
"""Send ``questions`` about ``batch_text`` through this side's client."""
return self._client.ask(conversation_tail_state(batch_text), questions)
return self._client.ask(self._judge_state(batch_text), questions)
def sharing_key(self, **dimensions: object) -> str:
"""Internal sharing identity (credential fingerprint, connection, limits, cache)."""
@ -175,7 +185,7 @@ class TypeSafeMemoryPrescreen:
return self.interpret(cached.answers, model=cached.model_for(self._questions), cached=True)
# No try/except: the transport failure is the coordinator's to record, and it
# writes no bucket here either way (a failure must not be remembered).
answer_set = self._client.ask(conversation_tail_state(request.batch_text), self._questions)
answer_set = self._client.ask(self._judge_state(request.batch_text), self._questions)
decision = self.interpret(answer_set.answers, model=answer_set.model, cached=False)
if decision is not None:
self._cache.put(request.digest, CachedVerdict(answers=dict(answer_set.answers), models={question_id: answer_set.model for question_id in answer_set.answers}))

View File

@ -0,0 +1,47 @@
"""OpenAI Decisions signal classifier: the same two additive hints over ``/v1/decisions``.
Like its pre-screen sibling, this subclass changes the endpoint and nothing else. The two
directions, their rubrics, ``hint_threshold``, the label mapping, the per-question cache buckets
and the rule that a model hint can never write a confirmation all stay in the base class — the
reinforcement evidence gate keeps reading the deterministic set only.
Configuration::
memory:
backend_config:
signal_classification:
mode: shadow
use: deerflow.agents.memory.signals.openai_decisions:OpenAIDecisionsSignalClassifier
config:
api_key_env: OPENAI_API_KEY
model: <the decision model this deployment names>
"""
from __future__ import annotations
from deerflow.agents.memory.signals.contract import CONFIGURATION_SOURCE
from deerflow.agents.memory.signals.typesafe import TypeSafeSignalClassifier
from deerflow.openai_decisions.client import OpenAIDecisionsClient, resolve_decisions_connection
from deerflow.typesafe.client import TransportFactory, TypeSafeClient
from deerflow.typesafe.connection import TypeSafeConnection
class OpenAIDecisionsSignalClassifier(TypeSafeSignalClassifier):
"""Map Decisions ``predicate`` probabilities for affirmation / negation onto hint labels."""
name = "openai_decisions"
policy_id = "deerflow.memory.signals.openai_decisions"
policy_version = "1.0.0"
def _resolve_connection(self, settings: dict[str, object]) -> TypeSafeConnection:
return resolve_decisions_connection(settings=settings, configuration_source=CONFIGURATION_SOURCE)
def _build_client(self, connection: TypeSafeConnection, *, transport_factory: TransportFactory | None) -> TypeSafeClient:
return OpenAIDecisionsClient(connection, transport_factory=transport_factory)
def _judge_state(self, batch_text: str) -> str:
"""The batch text itself: this endpoint's ``input`` is text, not a state object."""
return batch_text
__all__ = ["OpenAIDecisionsSignalClassifier"]

View File

@ -79,8 +79,8 @@ class TypeSafeSignalClassifier:
if mode not in MODES:
raise ValueError(f"memory.signal_classification.mode must be one of {list(MODES)}, got {mode!r}")
self._mode = mode
connection: TypeSafeConnection = resolve_connection(
settings={
self._connection = self._resolve_connection(
{
"api_key": api_key,
"api_key_env": api_key_env,
"base_url": base_url,
@ -89,12 +89,9 @@ class TypeSafeSignalClassifier:
"deadline_seconds": deadline_seconds,
"max_attempts": max_attempts,
"retry_backoff": retry_backoff,
},
defaults=typesafe_defaults(),
configuration_source=CONFIGURATION_SOURCE,
}
)
self._connection = connection
self._client = TypeSafeClient(connection, transport_factory=transport_factory)
self._client = self._build_client(self._connection, transport_factory=transport_factory)
self.hint_threshold = finite_float("hint_threshold", hint_threshold, minimum=0.0, maximum=1.0)
affirmation_criteria = _side_criteria(criteria, QUESTION_AFFIRMATION)
negation_criteria = _side_criteria(criteria, QUESTION_NEGATION)
@ -117,6 +114,19 @@ class TypeSafeSignalClassifier:
}
self._cache = AnswerCache(size=self.cache_size, ttl_seconds=self.cache_ttl_seconds)
# --- dialect hooks ------------------------------------------------------
def _resolve_connection(self, settings: dict[str, object]) -> TypeSafeConnection:
"""Resolve this provider's effective connection: consumer config, then its own defaults."""
return resolve_connection(settings=settings, defaults=typesafe_defaults(), configuration_source=CONFIGURATION_SOURCE)
def _build_client(self, connection: TypeSafeConnection, *, transport_factory: TransportFactory | None) -> TypeSafeClient:
return TypeSafeClient(connection, transport_factory=transport_factory)
def _judge_state(self, batch_text: str) -> object:
"""What the provider judges. A dialect whose endpoint takes plain input text overrides it."""
return conversation_tail_state(batch_text)
# --- side interface (shared with the coordinator) ---------------------
def questions(self) -> Mapping[str, Question]:
@ -125,7 +135,7 @@ class TypeSafeSignalClassifier:
def ask(self, batch_text: str, questions: Mapping[str, Question]):
"""Send ``questions`` about ``batch_text`` through this side's client."""
return self._client.ask(conversation_tail_state(batch_text), questions)
return self._client.ask(self._judge_state(batch_text), questions)
def sharing_key(self, **dimensions: object) -> str:
"""Internal sharing identity (credential fingerprint, connection, limits, cache)."""
@ -192,7 +202,7 @@ class TypeSafeSignalClassifier:
if not missing:
assert cached is not None
return self.interpret(answers, model=cached.model_for(self._questions), cached=True)
answer_set = self._client.ask(conversation_tail_state(request.batch_text), missing)
answer_set = self._client.ask(self._judge_state(request.batch_text), missing)
answers.update(answer_set.answers)
models.update({question_id: answer_set.model for question_id in answer_set.answers})
if answer_set.answers:

View File

@ -0,0 +1,71 @@
"""OpenAI Decisions guardrail provider: the same pre-execution risk gate, another endpoint.
Issue #7005 asked for a second decision backend behind the interfaces the Jev integration
already uses. Everything that makes this a *gate* is inherited from
:class:`~deerflow.guardrails.typesafe.TypeSafeGuardrailProvider` and deliberately not restated:
the strict-JSON local preflight, the character-count boundary, the threshold direction, the
cache, and fail-closed. Two copies of a security boundary drift apart, and the copy the primary
suite does not exercise is the one that goes quiet.
What differs is the wire, and only the wire:
* ``input`` is text, so the judged bytes are the canonical JSON the state digest already
identifies — the same bytes, not a second serialisation of them.
* a ``noul`` question becomes a ``predicate`` question, whose rubric rides in ``instructions``
(see ``deerflow.openai_decisions.client``).
* a refusal is an answer, not an HTTP failure; the inherited ``_recorded_answer`` already treats
"no validated answer for my one question" as ``TypeSafeGuardrailError`` with cause
``invalid_response``, so the gate denies on exactly the path it denies on today.
Configuration is the existing slot — no new schema::
guardrails:
provider:
use: deerflow.guardrails.openai_decisions:OpenAIDecisionsGuardrailProvider
config:
api_key_env: OPENAI_API_KEY
model: <the decision model this deployment names>
"""
from __future__ import annotations
from deerflow.guardrails.typesafe import TypeSafeGuardrailError, TypeSafeGuardrailProvider, _Probe
from deerflow.openai_decisions.client import OpenAIDecisionsClient, resolve_decisions_connection
from deerflow.typesafe.client import TransportFactory, TypeSafeClient
from deerflow.typesafe.connection import TypeSafeConnection
class OpenAIDecisionsGuardrailError(TypeSafeGuardrailError):
"""A Decisions evaluation could not produce a usable verdict.
Same family, same handling by ``guardrails.fail_closed``, different name so a log line and a
denial reason never attribute a decision to the wrong service.
"""
class OpenAIDecisionsGuardrailProvider(TypeSafeGuardrailProvider):
"""Deny tool calls whose Decisions ``predicate`` probability reaches ``threshold``.
Reads the same ``guardrails.provider.config`` block and the same optional fields as the Jev
provider; only ``base_url``/``model``/``api_key_env`` default differently, because they name a
different service.
"""
name = "openai_decisions"
policy_id = "deerflow.guardrails.openai_decisions"
policy_version = "1.0.0"
code_prefix = "openai_decisions"
provider_name = "OpenAI Decisions"
error_class = OpenAIDecisionsGuardrailError
def _resolve_connection(self, settings: dict[str, object]) -> TypeSafeConnection:
return resolve_decisions_connection(settings=settings, configuration_source=self.configuration_source)
def _build_client(self, connection: TypeSafeConnection, *, transport_factory: TransportFactory | None) -> TypeSafeClient:
return OpenAIDecisionsClient(connection, transport_factory=transport_factory)
def _state(self, probe: _Probe) -> str:
return probe.state_text
__all__ = ["OpenAIDecisionsGuardrailError", "OpenAIDecisionsGuardrailProvider"]

View File

@ -102,6 +102,12 @@ class _Probe:
tool_name: str
arguments_text: str
state_digest: str
state_text: str
"""The exact canonical JSON the digest was taken over.
Carried on the probe rather than re-derived: an endpoint dialect whose input is text instead
of a ``state`` object must judge the same bytes the digest identifies, and a second
serialisation of the same call could differ from the first."""
@dataclass(frozen=True)
@ -130,6 +136,21 @@ class TypeSafeGuardrailProvider:
policy_id = "deerflow.guardrails.typesafe"
policy_version = "1.1.0"
#: Prefix of every reason code this provider publishes. A dialect for another endpoint
#: overrides it so a denial names the service that actually decided.
code_prefix = "typesafe"
#: Names this service in a message about a response that carried no answer at all.
provider_name = "TypeSafe"
#: The config block whose values reached this provider, for the missing-credential message.
configuration_source = _CONFIGURATION_SOURCE
#: Raised when an evaluation produced no usable verdict. A dialect overrides the class so a
#: failure names its own service, while staying inside the same exception family the
#: middleware's ``guardrails.fail_closed`` mapping already handles.
error_class = TypeSafeGuardrailError
def __init__(
self,
*,
@ -151,8 +172,8 @@ class TypeSafeGuardrailProvider:
cache_ttl_seconds: float = 300.0,
transport_factory: TransportFactory | None = None,
) -> None:
connection: TypeSafeConnection = resolve_connection(
settings={
connection: TypeSafeConnection = self._resolve_connection(
{
"api_key": api_key,
"api_key_env": api_key_env,
"base_url": base_url,
@ -161,12 +182,10 @@ class TypeSafeGuardrailProvider:
"deadline_seconds": deadline_seconds,
"max_attempts": max_attempts,
"retry_backoff": retry_backoff,
},
defaults=typesafe_defaults(),
configuration_source=_CONFIGURATION_SOURCE,
}
)
self._connection = connection
self._client = TypeSafeClient(connection, transport_factory=transport_factory)
self._client = self._build_client(connection, transport_factory=transport_factory)
self._threshold = finite_float("threshold", threshold, minimum=0.0, maximum=1.0)
self._instructions = defaulted_text("instructions", instructions, DEFAULT_INSTRUCTIONS)
self._criteria_true = defaulted_text("criteria.true", criteria_entry(criteria, True), DEFAULT_CRITERIA_TRUE)
@ -184,6 +203,18 @@ class TypeSafeGuardrailProvider:
self._questions: dict[str, Question] = {_QUESTION_ID: self._question}
self._cache: OrderedDict[tuple[str, str], _CacheEntry] = OrderedDict()
# --- dialect hooks ------------------------------------------------------
def _code(self, suffix: str) -> str:
return f"{self.code_prefix}.{suffix}"
def _resolve_connection(self, settings: dict[str, object]) -> TypeSafeConnection:
"""Resolve this provider's effective connection: provider config, then its own defaults."""
return resolve_connection(settings=settings, defaults=typesafe_defaults(), configuration_source=self.configuration_source)
def _build_client(self, connection: TypeSafeConnection, *, transport_factory: TransportFactory | None) -> TypeSafeClient:
return TypeSafeClient(connection, transport_factory=transport_factory)
def release_policy_parameters(self) -> dict[str, object]:
"""Declare behaviour-affecting parameters for assembly identity (never the key).
@ -231,14 +262,14 @@ class TypeSafeGuardrailProvider:
try:
answer_set = self._client.ask(self._state(probe), self._questions)
except TypeSafeError as exc:
raise TypeSafeGuardrailError(str(exc), cause=exc.cause) from exc
raise self.error_class(str(exc), cause=exc.cause) from exc
return self._recorded_answer(answer_set)
async def _ask_async(self, probe: _Probe) -> _Answer:
try:
answer_set = await self._client.aask(self._state(probe), self._questions)
except TypeSafeError as exc:
raise TypeSafeGuardrailError(str(exc), cause=exc.cause) from exc
raise self.error_class(str(exc), cause=exc.cause) from exc
return self._recorded_answer(answer_set)
def _recorded_answer(self, answer_set: AnswerSet) -> _Answer:
@ -252,8 +283,8 @@ class TypeSafeGuardrailProvider:
answer = answer_set.noul(_QUESTION_ID)
if answer is None:
error = answer_set.errors_by_question.get(_QUESTION_ID)
message = error.message if error is not None else f"TypeSafe response has no answer for question {_QUESTION_ID!r}"
raise TypeSafeGuardrailError(message, cause=CAUSE_INVALID_RESPONSE)
message = error.message if error is not None else f"{self.provider_name} response has no answer for question {_QUESTION_ID!r}"
raise self.error_class(message, cause=CAUSE_INVALID_RESPONSE)
return _Answer(probability=answer.probability, model=recordable_model(answer_set.model))
def _state(self, probe: _Probe) -> dict[str, object]:
@ -273,14 +304,14 @@ class TypeSafeGuardrailProvider:
if self._allowed_tools is not None and tool_name not in self._allowed_tools:
return GuardrailDecision(
allow=False,
reasons=[GuardrailReason(code="typesafe.tool_not_allowed", message=f"typesafe.tool_not_allowed: tool={tool_name!r} not in configured allowed_tools policy={self._policy_ref()}")],
reasons=[GuardrailReason(code=self._code("tool_not_allowed"), message=f"{self._code('tool_not_allowed')}: tool={tool_name!r} not in configured allowed_tools policy={self._policy_ref()}")],
policy_id=self.policy_id,
metadata={"tool_not_allowed": True},
)
if self._tools is not None and tool_name not in self._tools:
return GuardrailDecision(
allow=True,
reasons=[GuardrailReason(code="typesafe.tool_not_probed", message=f"typesafe.tool_not_probed: tool={tool_name!r} not in configured tools policy={self._policy_ref()}")],
reasons=[GuardrailReason(code=self._code("tool_not_probed"), message=f"{self._code('tool_not_probed')}: tool={tool_name!r} not in configured tools policy={self._policy_ref()}")],
policy_id=self.policy_id,
metadata={"tool_not_probed": True},
)
@ -305,14 +336,14 @@ class TypeSafeGuardrailProvider:
# when a consumer asks it to; it changes no limit).
if len(arguments_text) > self._max_state_chars:
return self._state_unusable("too_large", limit=self._max_state_chars, length=len(arguments_text), digest=state_digest)
return _Probe(tool_name=tool_name, arguments_text=arguments_text, state_digest=state_digest)
return _Probe(tool_name=tool_name, arguments_text=arguments_text, state_digest=state_digest, state_text=state_text)
def _state_unusable(self, cause: str, **fields: object) -> GuardrailDecision:
"""Deny a call TypeSafe must not see: nothing truncated, nothing lossy is sent."""
detail = " ".join(f"{key}={value}" for key, value in fields.items())
return GuardrailDecision(
allow=False,
reasons=[GuardrailReason(code="typesafe.state_unusable", message=f"typesafe.state_unusable: cause={cause} {detail} policy={self._policy_ref()}")],
reasons=[GuardrailReason(code=self._code("state_unusable"), message=f"{self._code('state_unusable')}: cause={cause} {detail} policy={self._policy_ref()}")],
policy_id=self.policy_id,
metadata={"state_unusable_cause": cause, **fields},
)
@ -326,7 +357,7 @@ class TypeSafeGuardrailProvider:
return self._model_decision(allow=entry.allow, probability=entry.probability, model=entry.model, state_digest=entry.state_digest, cached=True)
def _model_decision(self, *, allow: bool, probability: float, model: str, state_digest: str, cached: bool) -> GuardrailDecision:
code = "typesafe.allowed" if allow else "typesafe.tool_call_risky"
code = self._code("allowed" if allow else "tool_call_risky")
# repr(): the recorded probability and threshold must replay to the same
# verdict, so neither is rounded for display. repr() round-trips exactly.
# ``model`` is already bounded by recordable_model: it is response

View File

@ -0,0 +1,45 @@
"""OpenAI Decisions transport for the host-side decision consumers.
Sibling of ``deerflow.typesafe`` for issue #7005: the same three consumer slots, the same
policy ownership, a different endpoint dialect.
* ``guardrails/openai_decisions.py`` — the pre-execution tool risk gate.
* ``agents/memory/prescreen/openai_decisions.py`` — the memory capture pre-screen.
* ``agents/memory/signals/openai_decisions.py`` — signal classification.
Shared here: the endpoint, the request envelope, question encoding, answer validation, and the
provider's own defaults. Transport lifecycle, authentication, retry, the deadline budget and
the bounded body read are inherited from ``deerflow.typesafe.client.TypeSafeClient`` rather than
restated, so one owner holds those rules for both backends.
The split of what stays in each adapter is `deerflow.typesafe`'s, unchanged: state content,
questions, criteria, thresholds, failure policy, cache semantics and business audit.
"""
from deerflow.openai_decisions.client import (
ANSWER_TYPE_REFUSAL,
CATEGORY_REFUSAL,
DEFAULT_API_KEY_ENV,
DEFAULT_BASE_URL,
ENDPOINT_PATH,
QUESTION_CHOICE,
QUESTION_PREDICATE,
OpenAIDecisionsClient,
encode_question,
predicate_instructions,
provider_defaults,
)
__all__ = [
"ANSWER_TYPE_REFUSAL",
"CATEGORY_REFUSAL",
"DEFAULT_API_KEY_ENV",
"DEFAULT_BASE_URL",
"ENDPOINT_PATH",
"OpenAIDecisionsClient",
"QUESTION_CHOICE",
"QUESTION_PREDICATE",
"encode_question",
"predicate_instructions",
"provider_defaults",
]

View File

@ -0,0 +1,222 @@
"""OpenAI Decisions transport: DeerFlow's question set spoken as one ``/v1/decisions`` request.
Issue #7005 asked for a second decision backend behind the interfaces the TypeSafe (Jev)
integration already uses, not a new decision system. The provider slots
(``guardrails.provider.use``, ``memory.prescreen.use``, ``memory.signal_classification.use``)
already resolve a class path and hand it the consumer's ``config``, so a new backend is a
sibling adapter and the workflows, thresholds and failure directions stay where they are.
Ownership follows `deerflow.typesafe`: this module owns the wire — endpoint, envelope, how a
question is encoded, how an answer is validated, and the error taxonomy — and nothing else.
Thresholds, rubrics, caches and the direction a failure takes stay in each adapter, so the
memory pre-screen still means "skip the extraction call" here exactly as it means there.
Transport itself (authentication, retry, backoff, the deadline budget, the bounded and
never-decoded body read) is inherited unchanged from
:class:`~deerflow.typesafe.client.TypeSafeClient`; re-declaring it in a second module would put
those security properties where they can drift.
Two differences are structural, and the mapping below exists because of them:
* **``input`` is the judged text, not a state object.** Jev takes a ``state`` mapping whose
shape each consumer owns. Here the consumer decides the same thing through its
``_judge_state`` hook, so nothing in this module knows what a tool call or a conversation
batch is.
* **A ``predicate`` question has no field for the two criteria.** A ``noul`` question's
``criteria`` say when the answer is true and when it is false; the Decisions endpoint only
carries ``instructions``. Folding them into the instruction text keeps one rubric per
question instead of silently dropping half of it, which would move the decision boundary
while the adapter went on reporting the same ``criteria`` in its policy identity.
A refusal is an *answer* on this endpoint (``{"type": "refusal"}``), not an HTTP failure.
It therefore becomes a question-level :class:`~deerflow.typesafe.client.QuestionError` — the
same channel a malformed answer uses — so the gate denies for want of a verdict while the
memory paths carry on, which is each consumer's existing policy rather than a new branch.
"""
from __future__ import annotations
import json
from collections.abc import Mapping
from deerflow.typesafe.client import (
CATEGORY_MISSING,
CATEGORY_TYPE,
Answer,
AnswerSet,
Question,
QuestionError,
TypeSafeClient,
validate_choice_label,
validate_probability,
)
from deerflow.typesafe.connection import TypeSafeConnection, resolve_connection
from deerflow.typesafe.errors import CAUSE_INVALID_RESPONSE, TypeSafeError
DEFAULT_BASE_URL = "https://api.openai.com"
DEFAULT_API_KEY_ENV = "OPENAI_API_KEY"
ENDPOINT_PATH = "/v1/decisions"
#: The Decisions spelling of a bounded yes/no judgement, which is what ``noul`` asks for.
QUESTION_PREDICATE = "predicate"
QUESTION_CHOICE = "choice"
#: ``refusal`` is a value of an answer's ``type``, so it is a per-question outcome.
ANSWER_TYPE_REFUSAL = "refusal"
CATEGORY_REFUSAL = "refusal"
def provider_defaults() -> dict[str, object]:
"""The fallback layer this backend passes to ``resolve_connection``.
``model`` is deliberately blank rather than a model id. Which decision model answers is an
operator choice with a cost and a data-handling implication, and a checked-in default would
send conversation and tool text to a model nobody selected. The resolver rejects a blank
model at construction, so a deployment naming this backend has to name a model too.
"""
return {"base_url": DEFAULT_BASE_URL, "api_key_env": DEFAULT_API_KEY_ENV, "model": ""}
def resolve_decisions_connection(*, settings: Mapping[str, object] | None = None, defaults: Mapping[str, object] | None = None, configuration_source: str) -> TypeSafeConnection:
"""Resolve a connection that speaks this endpoint, or raise at construction.
One owner pairs the provider's defaults with its request path: a "Decisions" connection
resolved without ``ENDPOINT_PATH`` would post to the other service's URL, and nothing but a
live request would say so. ``defaults`` is layered *over* the provider's own defaults, which
are in turn overridden by ``settings`` — the precedence each adapter already documents.
"""
return resolve_connection(
settings=settings,
defaults={**provider_defaults(), **(defaults or {})},
configuration_source=configuration_source,
endpoint_path=ENDPOINT_PATH,
provider_label="OpenAI Decisions",
)
class OpenAIDecisionsClient(TypeSafeClient):
"""One consumer's Decisions client, built from its effective connection.
Inherits the whole transport lifecycle; overrides only the request skeleton and the
response parsing, which are the parts that are protocol-specific.
"""
#: ``noul``/System One wording in an inherited message would attribute a decision to the
#: wrong service, and these messages reach logs, denial reasons and the evaluation report.
provider_label = "OpenAI Decisions"
def _payload(self, state: object, questions: Mapping[str, Question]) -> dict[str, object]:
return {
"model": self._connection.model,
"input": state,
"questions": [encode_question(question_id, question) for question_id, question in questions.items()],
}
def _parse(self, body: bytes, questions: Mapping[str, Question]) -> AnswerSet:
"""Split one response into an envelope plus per-question outcomes, as ``typesafe`` does."""
try:
payload = json.loads(body)
# ``RecursionError`` and not ``ValueError`` is what deep nesting raises, and it must stay
# inside this taxonomy: a hostile or broken endpoint must not escape as an unrelated
# exception, which the consumer would lose the round's record over.
except (ValueError, RecursionError) as exc:
raise TypeSafeError(f"{self.provider_label} response was not valid JSON", cause=CAUSE_INVALID_RESPONSE) from exc
if not isinstance(payload, dict):
raise TypeSafeError(f"{self.provider_label} response must be a JSON object", cause=CAUSE_INVALID_RESPONSE)
raw_answers = payload.get("answers")
if not isinstance(raw_answers, list):
raise TypeSafeError(f"{self.provider_label} response is missing an 'answers' array", cause=CAUSE_INVALID_RESPONSE)
by_name = _answers_by_name(raw_answers)
# ``model`` is optional here: the endpoint may echo the served version, and when it does
# that value is what a consumer records. When it does not, the model this request named is
# recorded instead — provenance is still needed for the audit trail, and unlike a value
# the endpoint returned it cannot have been chosen to be echoed into a log.
model = payload.get("model")
served = model if isinstance(model, str) and model else self._connection.model
validated: dict[str, Answer] = {}
errors: dict[str, QuestionError] = {}
for question_id, question in questions.items():
answer, error = validate_answer(question_id, question, by_name.get(question_id), provider_label=self.provider_label)
if error is not None:
errors[question_id] = error
elif answer is not None:
validated[question_id] = answer
return AnswerSet(model=served, answers=validated, errors_by_question=errors)
def encode_question(question_id: str, question: Question) -> dict[str, object]:
"""Encode one shared :class:`Question` as a Decisions question object."""
if question.type == QUESTION_CHOICE:
# A choice question does carry its options, so the criteria map onto them directly and the
# instruction text is left alone; silently narrowing it would change what is judged.
return {
"type": QUESTION_CHOICE,
"name": question_id,
"instructions": question.instructions,
"choices": [{"value": label, "description": description} for label, description in (question.criteria or {}).items()],
}
return {"type": QUESTION_PREDICATE, "name": question_id, "instructions": predicate_instructions(question)}
def predicate_instructions(question: Question) -> str:
"""The rubric a predicate question has no field for, folded into its instructions."""
if not question.criteria:
return question.instructions
rubric = "".join(f"\nAnswer {label} when: {description}" for label, description in question.criteria.items())
return f"{question.instructions}{rubric}"
def _answers_by_name(raw_answers: list[object]) -> dict[str, dict[object, object]]:
"""Index the answer array by ``name``, first occurrence winning.
An array can repeat a name while a mapping cannot, and which of two conflicting answers a
consumer would act on must not depend on how a server orders them. Answers without a usable
name, and names nobody asked about, are ignored rather than reported: the shared contract is
that only the questions actually asked appear in the outcome.
"""
by_name: dict[str, dict[object, object]] = {}
for entry in raw_answers:
if not isinstance(entry, dict):
continue
name = entry.get("name")
if isinstance(name, str) and name not in by_name:
by_name[name] = entry
return by_name
def validate_answer(question_id: str, question: Question, entry: dict[object, object] | None, *, provider_label: str) -> tuple[Answer | None, QuestionError | None]:
"""Validate one answer against the question that was asked, reusing the shared rules."""
if entry is None:
return None, QuestionError(question_id=question_id, category=CATEGORY_MISSING, message=f"{provider_label} response has no answer for question {question_id!r}")
answer_type = entry.get("type")
if answer_type == ANSWER_TYPE_REFUSAL:
# Only the outcome is reported. A refusal object is response content and can carry text the
# endpoint chose, which must not reach a log or a reason message unchecked.
return None, QuestionError(question_id=question_id, category=CATEGORY_REFUSAL, message=f"{provider_label} refused to answer question {question_id!r}")
expected = QUESTION_CHOICE if question.type == QUESTION_CHOICE else QUESTION_PREDICATE
if answer_type != expected:
return None, QuestionError(
question_id=question_id,
category=CATEGORY_TYPE,
message=f"{provider_label} answer for {question_id!r} is not a {expected} answer (type={type(answer_type).__name__})",
)
if expected == QUESTION_PREDICATE:
return validate_probability(question_id, entry.get("probability"), provider_label=provider_label, field_name="predicate")
return validate_choice_label(question_id, question, entry.get("choice"), provider_label=provider_label)
__all__ = [
"ANSWER_TYPE_REFUSAL",
"CATEGORY_REFUSAL",
"DEFAULT_API_KEY_ENV",
"DEFAULT_BASE_URL",
"ENDPOINT_PATH",
"QUESTION_CHOICE",
"QUESTION_PREDICATE",
"OpenAIDecisionsClient",
"encode_question",
"predicate_instructions",
"provider_defaults",
"resolve_decisions_connection",
]

View File

@ -50,3 +50,22 @@ Credential resolution is per *layer* too — the first layer that sets `api_key`
consumer naming an environment variable is never silently switched to another
layer's key. Likewise `base_url` must be an `http(s)` URL with no query, fragment
or embedded credentials, checked at construction rather than on every request.
**Dialects.** `deerflow/openai_decisions/` (issue #7005) speaks a second endpoint over this same
client rather than copying it, because the rules above are security properties and a second copy
of them can drift unnoticed. Two seams carry a dialect, and they are the only two:
``TypeSafeClient.provider_label``, which names the service in every request-level message (a
failure must not tell an operator about a service they did not configure), and
``TypeSafeConnection.endpoint_path``, which the connection composes into ``url`` so an endpoint
path belongs to the backend that resolves the connection rather than to the base URL it happens
to share. The endpoint path is included in ``public_parameters()`` and therefore in both the
sharing key and public policy identity: different dialects behind one gateway cannot share a
combined memory request. Matching endpoints may still combine when the other settings match.
``resolve_connection(provider_label=...)`` also names the configured service in credential
and model construction errors. ``openai_decisions.client.resolve_decisions_connection`` pairs
the endpoint and service label, so a
"Decisions" connection cannot be built pointing at System One. A dialect adds no threshold, no
rubric and no failure policy: the three adapters override their connection resolution, their
client and the value the endpoint judges (``input`` text instead of a ``state`` object), and
inherit the rest.

View File

@ -172,7 +172,7 @@ type Answer = NoulAnswer | ChoiceAnswer
class QuestionError:
"""A question-level failure: the request was usable, this question's answer was not.
``category`` is one of ``missing``, ``type``, ``probability`` or ``label``.
``category`` is one of ``missing``, ``type``, ``probability``, ``label`` or ``refusal``.
``message`` names the problem without repeating the response body, which a
malformed or hostile endpoint can fill with echoed state.
"""
@ -217,6 +217,11 @@ class TypeSafeClient:
§3.1).
"""
#: Names the endpoint in every request-level message this client raises. A
#: dialect subclass that speaks another provider's protocol overrides it, so a
#: failure never reports the wrong operator's service.
provider_label = "TypeSafe"
def __init__(self, connection: TypeSafeConnection, *, transport_factory: TransportFactory | None = None) -> None:
self._connection = connection
self._transport_factory = transport_factory
@ -333,9 +338,9 @@ class TypeSafeClient:
# request it failed to build — for a rejected header that is the whole
# ``Bearer <key>`` value — and the guardrail's ``logger.exception``
# prints the cause chain. The type name below is kept for diagnosis.
raise _RetryableAttempt(f"TypeSafe request failed: {type(exc).__name__}", cause=CAUSE_TRANSPORT) from None
raise _RetryableAttempt(f"{self.provider_label} request failed: {type(exc).__name__}", cause=CAUSE_TRANSPORT) from None
except httpx.HTTPError as exc:
raise TypeSafeError(f"TypeSafe request failed: {type(exc).__name__}", cause=CAUSE_TRANSPORT) from None
raise TypeSafeError(f"{self.provider_label} request failed: {type(exc).__name__}", cause=CAUSE_TRANSPORT) from None
async def _attempt_async(self, client: httpx.AsyncClient, payload: dict[str, object], headers: dict[str, str], questions: Mapping[str, Question], deadline_at: float) -> AnswerSet:
try:
@ -352,9 +357,9 @@ class TypeSafeClient:
except httpx.TransportError as exc:
# ``from None`` for the same reason as the sync path: a rejected header
# would otherwise carry the ``Bearer`` value in the printed chain.
raise _RetryableAttempt(f"TypeSafe request failed: {type(exc).__name__}", cause=CAUSE_TRANSPORT) from None
raise _RetryableAttempt(f"{self.provider_label} request failed: {type(exc).__name__}", cause=CAUSE_TRANSPORT) from None
except httpx.HTTPError as exc:
raise TypeSafeError(f"TypeSafe request failed: {type(exc).__name__}", cause=CAUSE_TRANSPORT) from None
raise TypeSafeError(f"{self.provider_label} request failed: {type(exc).__name__}", cause=CAUSE_TRANSPORT) from None
def _status_error(self, status_code: int) -> TypeSafeError:
# Only the numeric status is reported. The response body *and* the HTTP
@ -362,7 +367,7 @@ class TypeSafeClient:
# (raw tool arguments, conversation text) in either, and this message
# reaches middleware logs and the evaluation report.
hint = " (check the API key configured via api_key / the API key environment variable)" if status_code == _UNAUTHORIZED_STATUS else ""
message = f"TypeSafe returned HTTP {status_code}{hint}"
message = f"{self.provider_label} returned HTTP {status_code}{hint}"
if status_code in _RETRYABLE_STATUS_CODES:
return _RetryableAttempt(message, cause=CAUSE_HTTP_STATUS)
return TypeSafeError(message, cause=CAUSE_HTTP_STATUS)
@ -378,15 +383,15 @@ class TypeSafeClient:
# would escape this module's taxonomy and the consumer would lose the round's
# record instead of falling back.
except (ValueError, RecursionError) as exc:
raise TypeSafeError("TypeSafe response was not valid JSON", cause=CAUSE_INVALID_RESPONSE) from exc
raise TypeSafeError(f"{self.provider_label} response was not valid JSON", cause=CAUSE_INVALID_RESPONSE) from exc
if not isinstance(payload, dict):
raise TypeSafeError("TypeSafe response must be a JSON object", cause=CAUSE_INVALID_RESPONSE)
raise TypeSafeError(f"{self.provider_label} response must be a JSON object", cause=CAUSE_INVALID_RESPONSE)
model = payload.get("model")
if not isinstance(model, str) or not model:
raise TypeSafeError("TypeSafe response is missing a non-empty 'model'", cause=CAUSE_INVALID_RESPONSE)
raise TypeSafeError(f"{self.provider_label} response is missing a non-empty 'model'", cause=CAUSE_INVALID_RESPONSE)
answers = payload.get("answers")
if not isinstance(answers, dict):
raise TypeSafeError("TypeSafe response is missing an 'answers' object", cause=CAUSE_INVALID_RESPONSE)
raise TypeSafeError(f"{self.provider_label} response is missing an 'answers' object", cause=CAUSE_INVALID_RESPONSE)
validated: dict[str, Answer] = {}
errors: dict[str, QuestionError] = {}
@ -406,7 +411,7 @@ class TypeSafeClient:
raise self._deadline_error()
def _deadline_error(self) -> TypeSafeError:
return TypeSafeError(f"TypeSafe evaluation exceeded deadline_seconds={self._connection.deadline_seconds:g}", cause=CAUSE_DEADLINE)
return TypeSafeError(f"{self.provider_label} evaluation exceeded deadline_seconds={self._connection.deadline_seconds:g}", cause=CAUSE_DEADLINE)
def _remaining(self, deadline_at: float) -> float:
remaining = deadline_at - time.monotonic()
@ -491,32 +496,49 @@ def _validate_answer(question_id: str, question: Question, raw: object) -> tuple
return _validate_choice(question_id, question, raw)
def _validate_noul(question_id: str, raw: dict[object, object]) -> tuple[Answer | None, QuestionError | None]:
raw_probability = raw.get("noul")
# bool is an int subclass, and json.loads parses NaN/Infinity literals, so both
# are rejected here rather than reaching a consumer's comparison. Only the
# offending type is reported, for the reason in _validate_answer.
def validate_probability(question_id: str, raw_probability: object, *, provider_label: str, field_name: str) -> tuple[Answer | None, QuestionError | None]:
"""One owner for the probability rules every dialect's numeric answer needs.
``bool`` is an ``int`` subclass and ``json.loads`` accepts the ``NaN``/``Infinity``
literals, so both are rejected here rather than reaching a consumer's threshold
comparison. Only the offending type is reported, for the reason in
:func:`_validate_answer`. A ``QuestionError`` is returned instead of raising so a
bad answer cannot discard the other valid answers in the same response (design §2.3).
"""
if isinstance(raw_probability, bool) or not isinstance(raw_probability, (int, float)):
return None, QuestionError(question_id=question_id, category=CATEGORY_PROBABILITY, message=f"TypeSafe returned a non-numeric noul probability (type={type(raw_probability).__name__})")
return None, QuestionError(question_id=question_id, category=CATEGORY_PROBABILITY, message=f"{provider_label} returned a non-numeric {field_name} probability (type={type(raw_probability).__name__})")
try:
probability = float(raw_probability)
except OverflowError:
# A JSON integer too large for float() (e.g. 10**400) must stay a
# per-question error: letting it escape would discard the other valid
# answers in the same response (design §2.3).
return None, QuestionError(question_id=question_id, category=CATEGORY_PROBABILITY, message="TypeSafe returned a noul probability outside [0, 1]")
return None, QuestionError(question_id=question_id, category=CATEGORY_PROBABILITY, message=f"{provider_label} returned a {field_name} probability outside [0, 1]")
if not math.isfinite(probability) or not 0.0 <= probability <= 1.0:
return None, QuestionError(question_id=question_id, category=CATEGORY_PROBABILITY, message="TypeSafe returned a noul probability outside [0, 1]")
return None, QuestionError(question_id=question_id, category=CATEGORY_PROBABILITY, message=f"{provider_label} returned a {field_name} probability outside [0, 1]")
return NoulAnswer(probability=probability), None
def _validate_noul(question_id: str, raw: dict[object, object]) -> tuple[Answer | None, QuestionError | None]:
return validate_probability(question_id, raw.get("noul"), provider_label="TypeSafe", field_name="noul")
def validate_choice_label(question_id: str, question: Question, raw_label: object, *, provider_label: str) -> tuple[Answer | None, QuestionError | None]:
"""One owner for the rule that a ``choice`` answer must be one of the labels asked for.
The label set is the question's own criteria: an answer outside it is a
question-level failure, never a verdict, because a consumer maps labels to
behaviour and an unrecognised one has no mapping to apply.
"""
if not isinstance(raw_label, str) or not raw_label:
return None, QuestionError(question_id=question_id, category=CATEGORY_LABEL, message=f"{provider_label} returned a non-textual choice label (type={type(raw_label).__name__})")
if question.criteria is None or raw_label not in question.criteria:
return None, QuestionError(question_id=question_id, category=CATEGORY_LABEL, message=f"{provider_label} returned a choice label outside the configured criteria")
return ChoiceAnswer(label=raw_label), None
def _validate_choice(question_id: str, question: Question, raw: dict[object, object]) -> tuple[Answer | None, QuestionError | None]:
label = raw.get("choice")
if not isinstance(label, str) or not label:
return None, QuestionError(question_id=question_id, category=CATEGORY_LABEL, message=f"TypeSafe returned a non-textual choice label (type={type(label).__name__})")
if question.criteria is None or label not in question.criteria:
return None, QuestionError(question_id=question_id, category=CATEGORY_LABEL, message="TypeSafe returned a choice label outside the configured criteria")
return ChoiceAnswer(label=label), None
return validate_choice_label(question_id, question, raw.get("choice"), provider_label="TypeSafe")
def _factory_identity(factory: TransportFactory | None) -> str | None:
@ -552,5 +574,7 @@ __all__ = [
"TransportFactory",
"TypeSafeClient",
"recordable_model",
"validate_choice_label",
"validate_probability",
"wire_size",
]

View File

@ -81,11 +81,15 @@ class TypeSafeConnection:
deadline_seconds: float
max_attempts: int
retry_backoff: float
#: Which request path the endpoint serves. It belongs to the *dialect*, not the operator:
#: a connection whose base URL points at another service must not report a System One URL, so
#: every consumer resolves it through the backend it is about to call.
endpoint_path: str = ENDPOINT_PATH
@property
def url(self) -> str:
"""The System One endpoint this connection posts to."""
return f"{self.base_url.rstrip('/')}{ENDPOINT_PATH}"
"""The endpoint this connection posts to."""
return f"{self.base_url.rstrip('/')}{self.endpoint_path}"
def credential_fingerprint(self) -> str:
"""A short digest of the credential, to compare consumers without recording either key."""
@ -96,6 +100,7 @@ class TypeSafeConnection:
return {
"model": self.model,
"base_url": self.base_url.rstrip("/"),
"endpoint_path": self.endpoint_path,
"timeout": self.timeout,
"deadline_seconds": self.deadline_seconds,
"max_attempts": self.max_attempts,
@ -108,6 +113,8 @@ def resolve_connection(
settings: Mapping[str, object] | None = None,
defaults: Mapping[str, object] | None = None,
configuration_source: str,
endpoint_path: str = ENDPOINT_PATH,
provider_label: str = "TypeSafe",
) -> TypeSafeConnection:
"""Resolve the effective connection for one consumer.
@ -122,7 +129,13 @@ def resolve_connection(
``configuration_source`` names the consumer's settings location (for example
``guardrails.provider.config``); it reaches the missing-credential message so
an operator knows which block to edit. It never carries the credential.
``endpoint_path`` is what the calling backend serves, not something a consumer
configures; it is validated here so a malformed value cannot become a per-request URL error.
``provider_label`` names that backend in construction-time diagnostics.
"""
if not isinstance(endpoint_path, str) or not endpoint_path.startswith("/"):
raise ValueError("endpoint_path must be an absolute request path such as '/v1/systemone'")
override = _configured(settings)
fallback = _configured(defaults)
@ -132,7 +145,7 @@ def resolve_connection(
api_key = _resolve_credential(override, fallback, api_key_env)
if not isinstance(api_key, str) or not api_key:
raise ValueError(f"TypeSafe requires an API key: pass 'api_key' in {configuration_source} or set the {api_key_env} environment variable")
raise ValueError(f"{provider_label} requires an API key: pass 'api_key' in {configuration_source} or set the {api_key_env} environment variable")
# Shape only, and never echoed: a key that cannot travel as a header value is a
# configuration error here instead of a per-call protocol error whose message
# carries the whole ``Bearer`` header into the guardrail's exception log.
@ -142,7 +155,7 @@ def resolve_connection(
model = _first(override, fallback, "model", DEFAULT_MODEL)
if not isinstance(model, str) or not model.strip():
raise ValueError("model must be a non-empty string; TypeSafe returns the version it actually served")
raise ValueError(f"{provider_label} model must be a non-empty string")
return TypeSafeConnection(
api_key=api_key,
@ -153,6 +166,7 @@ def resolve_connection(
deadline_seconds=finite_float("deadline_seconds", _first(override, fallback, "deadline_seconds", DEFAULT_DEADLINE_SECONDS), minimum=0.0, exclusive=True),
max_attempts=whole_number("max_attempts", _first(override, fallback, "max_attempts", DEFAULT_MAX_ATTEMPTS), minimum=1),
retry_backoff=finite_float("retry_backoff", _first(override, fallback, "retry_backoff", DEFAULT_RETRY_BACKOFF), minimum=0.0),
endpoint_path=endpoint_path,
)
@ -162,6 +176,7 @@ def resolve_connection_for_mode(
settings: Mapping[str, object] | None = None,
defaults: Mapping[str, object] | None = None,
configuration_source: str,
endpoint_path: str = ENDPOINT_PATH,
) -> TypeSafeConnection | None:
"""Resolve a mode-bearing consumer's connection, or ``None`` when it is off.
@ -175,7 +190,7 @@ def resolve_connection_for_mode(
"""
if mode == MODE_OFF:
return None
return resolve_connection(settings=settings, defaults=defaults, configuration_source=configuration_source)
return resolve_connection(settings=settings, defaults=defaults, configuration_source=configuration_source, endpoint_path=endpoint_path)
def _configured(settings: Mapping[str, object] | None) -> dict[str, object]:

View File

@ -0,0 +1,552 @@
"""Tests for the OpenAI Decisions backend (issue #7005).
Three things are pinned here, and each is a claim the change could silently break:
* **The wire mapping both directions.** A ``noul`` question must leave as a ``predicate``
question whose rubric still travels with it, and an answer array must come back as the same
``NoulAnswer`` / ``ChoiceAnswer`` / ``QuestionError`` shapes the Jev client produces — because
every consumer's threshold, cache bucket and failure direction is written against those.
* **That the shared transport rules still hold for a second dialect.** Bounded never-decoded
body, the two-layer error split, the identity encoding, and "a refusal is a per-question
outcome". A duplicated transport would have had to re-earn each of those.
* **That the gate's direction did not move.** The tool gate still denies at or above its
threshold and fails closed on a refusal, while the memory sides still fall back — the same
request, opposite policies, owned by each adapter as before.
Everything runs offline through ``httpx.MockTransport``, the pattern the Jev suites use.
"""
from __future__ import annotations
import gzip
import hashlib
import json
import httpx
import pytest
from deerflow.agents.memory.prescreen.contract import MODE_ENFORCE, resolve_memory_prescreen
from deerflow.agents.memory.prescreen.openai_decisions import OpenAIDecisionsMemoryPrescreen
from deerflow.agents.memory.prescreen.typesafe import QUESTION_ID as PRESCREEN_QUESTION
from deerflow.agents.memory.prescreen.typesafe import TypeSafeMemoryPrescreen
from deerflow.agents.memory.signals.contract import LABEL_REINFORCEMENT
from deerflow.agents.memory.signals.coordinator import MemoryBatchContext, MemorySignalCoordinator
from deerflow.agents.memory.signals.openai_decisions import OpenAIDecisionsSignalClassifier
from deerflow.agents.memory.signals.typesafe import QUESTION_AFFIRMATION, QUESTION_NEGATION, TypeSafeSignalClassifier
from deerflow.guardrails.openai_decisions import OpenAIDecisionsGuardrailError, OpenAIDecisionsGuardrailProvider
from deerflow.guardrails.provider import GuardrailRequest
from deerflow.openai_decisions.client import (
CATEGORY_REFUSAL,
DEFAULT_API_KEY_ENV,
DEFAULT_BASE_URL,
ENDPOINT_PATH,
QUESTION_CHOICE,
QUESTION_PREDICATE,
OpenAIDecisionsClient,
resolve_decisions_connection,
)
from deerflow.typesafe.client import (
CATEGORY_LABEL,
CATEGORY_MISSING,
CATEGORY_PROBABILITY,
CATEGORY_TYPE,
MAX_RESPONSE_BYTES,
QUESTION_NOUL,
NoulAnswer,
Question,
)
from deerflow.typesafe.errors import CAUSE_HTTP_STATUS, CAUSE_INVALID_RESPONSE, TypeSafeError
_API_KEY = "decisions-test-key"
_MODEL = "gpt-decisions-test"
_RUBRIC_TRUE = "it deletes data"
_RUBRIC_FALSE = "it only reads"
_PREDICATE = Question(type=QUESTION_NOUL, instructions="Is this risky?", criteria={"true": _RUBRIC_TRUE, "false": _RUBRIC_FALSE})
_CHOICE = Question(type=QUESTION_CHOICE, instructions="Which label?", criteria={"keep": "worth keeping", "drop": "not worth keeping"})
#: The canonical state JSON the gate digests, written out by hand rather than re-derived by the
#: code under test: the assertion below would pass even if the serialisation changed shape.
_EXPECTED_GATE_INPUT = '{"tool_call":{"arguments":"{\\"command\\":\\"ls\\"}","name":"bash"}}'
class _Server:
"""Fake Decisions endpoint that records every request it receives."""
def __init__(self, responder=None) -> None:
self.requests: list[httpx.Request] = []
self._responder = responder or (lambda request: httpx.Response(200, json={"answers": [_predicate_answer("q", 0.9)]}))
def handle(self, request: httpx.Request) -> httpx.Response:
self.requests.append(request)
return self._responder(request)
def transport(self) -> httpx.MockTransport:
return httpx.MockTransport(self.handle)
@property
def body(self) -> dict:
return json.loads(self.requests[-1].content.decode("utf-8"))
def _predicate_probability_body(name: str, probability_json: str) -> bytes:
return f'{{"answers":[{{"type":"{QUESTION_PREDICATE}","name":"{name}","probability":{probability_json}}}]}}'.encode()
def _predicate_answer(name: str, probability: float) -> dict:
return {"type": QUESTION_PREDICATE, "name": name, "probability": probability}
def _client(server: _Server, **settings) -> OpenAIDecisionsClient:
connection = resolve_decisions_connection(settings={"api_key": _API_KEY, "model": _MODEL, **settings}, configuration_source="tests.openai_decisions")
return OpenAIDecisionsClient(connection, transport_factory=server.transport)
def _ask(server: _Server, questions: dict[str, Question], input_value: object = "judged text") -> object:
return _client(server).ask(input_value, questions)
# --- request mapping --------------------------------------------------------
class TestRequestMapping:
def test_a_noul_question_leaves_as_a_predicate_question(self):
server = _Server()
_ask(server, {"q": _PREDICATE})
question = server.body["questions"][0]
assert question["type"] == QUESTION_PREDICATE
assert question["name"] == "q"
def test_the_rubric_travels_because_a_predicate_question_has_no_field_for_it(self):
"""Dropping the criteria would move the decision boundary while the adapter kept reporting
the same ``criteria`` in its policy identity."""
server = _Server()
_ask(server, {"q": _PREDICATE})
instructions = server.body["questions"][0]["instructions"]
assert instructions.startswith("Is this risky?")
assert _RUBRIC_TRUE in instructions
assert _RUBRIC_FALSE in instructions
def test_a_choice_question_keeps_its_instructions_and_carries_its_options(self):
server = _Server(lambda request: httpx.Response(200, json={"answers": [{"type": QUESTION_CHOICE, "name": "q", "choice": "keep", "confidence": 0.9}]}))
_ask(server, {"q": _CHOICE})
question = server.body["questions"][0]
assert question == {
"type": QUESTION_CHOICE,
"name": "q",
"instructions": "Which label?",
"choices": [{"value": "keep", "description": "worth keeping"}, {"value": "drop", "description": "not worth keeping"}],
}
def test_the_judged_value_is_input_and_no_state_envelope_is_sent(self):
server = _Server()
_ask(server, {"q": _PREDICATE}, input_value="the batch text")
assert server.body["input"] == "the batch text"
assert "state" not in server.body
assert server.body["model"] == _MODEL
def test_it_posts_to_the_decisions_endpoint_with_the_shared_headers(self):
server = _Server()
_ask(server, {"q": _PREDICATE})
request = server.requests[-1]
assert request.url.path == ENDPOINT_PATH
assert request.method == "POST"
assert request.headers["authorization"] == f"Bearer {_API_KEY}"
# The bounded body read depends on this: an encoded body would make the *decoded* size the
# number that matters, and the client would be counting the wrong thing.
assert request.headers["accept-encoding"] == "identity"
def test_questions_keep_their_request_order(self):
server = _Server()
_ask(server, {QUESTION_NEGATION: _PREDICATE, QUESTION_AFFIRMATION: _PREDICATE})
assert [entry["name"] for entry in server.body["questions"]] == [QUESTION_NEGATION, QUESTION_AFFIRMATION]
class TestConnectionDefaults:
@pytest.mark.parametrize("provider", [OpenAIDecisionsGuardrailProvider, OpenAIDecisionsMemoryPrescreen, OpenAIDecisionsSignalClassifier])
@pytest.mark.parametrize("settings, field", [({"api_key": "", "model": _MODEL}, "API key"), ({"api_key": _API_KEY}, "model")])
def test_construction_errors_name_the_configured_service(self, provider, settings, field):
with pytest.raises(ValueError) as excinfo:
provider(**settings)
message = str(excinfo.value)
assert "OpenAI Decisions" in message
assert "TypeSafe" not in message
assert field in message
assert _API_KEY not in message
def test_the_backend_defaults_to_its_own_service_not_the_jev_one(self):
server = _Server()
client = _client(server)
assert client.connection.base_url == DEFAULT_BASE_URL
assert client.connection.url == f"{DEFAULT_BASE_URL}{ENDPOINT_PATH}"
assert client.connection.api_key_env == DEFAULT_API_KEY_ENV
def test_a_deployment_must_name_its_model_rather_than_inheriting_another_services_default(self):
"""No checked-in model id: an empty default fails loudly at construction."""
with pytest.raises(ValueError, match="model must be a non-empty string"):
resolve_decisions_connection(settings={"api_key": _API_KEY}, configuration_source="tests.openai_decisions")
def test_an_operator_override_still_wins(self):
server = _Server()
client = _client(server, base_url="http://127.0.0.1:8000", model="local-decisions")
assert client.connection.url == "http://127.0.0.1:8000/v1/decisions"
# --- response mapping ------------------------------------------------------
class TestResponseMapping:
def test_a_probability_comes_back_as_the_shared_answer_type(self):
server = _Server(lambda request: httpx.Response(200, json={"model": "served-1.2", "answers": [_predicate_answer("q", 0.31)]}))
answers = _ask(server, {"q": _PREDICATE})
assert answers.answers["q"] == NoulAnswer(probability=0.31)
assert answers.model == "served-1.2"
def test_a_response_without_a_model_records_the_one_that_was_asked_for(self):
"""Provenance is still needed for the audit trail, and the requested model is the only
value here that the endpoint did not choose."""
server = _Server(lambda request: httpx.Response(200, json={"answers": [_predicate_answer("q", 0.31)]}))
answers = _ask(server, {"q": _PREDICATE})
assert answers.model == _MODEL
def test_a_refusal_is_a_per_question_outcome_not_a_request_failure(self):
server = _Server(lambda request: httpx.Response(200, json={"answers": [{"type": "refusal", "name": "q"}]}))
answers = _ask(server, {"q": _PREDICATE})
assert answers.answers == {}
error = answers.errors_by_question["q"]
assert error.category == CATEGORY_REFUSAL
assert "refused to answer" in error.message
def test_a_valid_answer_survives_an_invalid_one_in_the_same_array(self):
server = _Server(lambda request: httpx.Response(200, json={"answers": [_predicate_answer("good", 0.4), {"type": QUESTION_PREDICATE, "name": "bad", "probability": "0.9"}]}))
answers = _ask(server, {"good": _PREDICATE, "bad": _PREDICATE})
assert answers.answers["good"] == NoulAnswer(probability=0.4)
assert answers.errors_by_question["bad"].category == CATEGORY_PROBABILITY
@pytest.mark.parametrize(
("probability_json", "expected"),
[
("true", "non-numeric"), # bool is an int subclass, so True would judge as 1.0
('"0.9"', "non-numeric"), # a JSON string must not silently become a number
("1.5", "outside [0, 1]"),
("-0.1", "outside [0, 1]"),
("Infinity", "outside [0, 1]"), # a bare literal json.loads accepts and a threshold cannot
],
)
def test_the_shared_numeric_rules_apply_to_this_dialect_too(self, probability_json, expected):
"""Written as raw JSON text rather than a Python value, because the payload under test is
what an endpoint can actually put on the wire."""
server = _Server(lambda request: httpx.Response(200, content=_predicate_probability_body("q", probability_json), headers={"content-type": "application/json"}))
answers = _ask(server, {"q": _PREDICATE})
error = answers.errors_by_question["q"]
assert error.category == CATEGORY_PROBABILITY
assert expected in error.message
assert probability_json not in error.message, "the response body is never repeated into a message"
def test_a_probability_answered_to_the_wrong_question_kind_is_not_a_verdict(self):
server = _Server(lambda request: httpx.Response(200, json={"answers": [{"type": "score", "name": "q", "score": 3}]}))
answers = _ask(server, {"q": _PREDICATE})
assert answers.errors_by_question["q"].category == CATEGORY_TYPE
def test_a_choice_label_outside_the_asked_set_is_rejected(self):
server = _Server(lambda request: httpx.Response(200, json={"answers": [{"type": QUESTION_CHOICE, "name": "q", "choice": "shred"}]}))
answers = _ask(server, {"q": _CHOICE})
assert answers.errors_by_question["q"].category == CATEGORY_LABEL
def test_the_first_answer_for_a_repeated_name_wins(self):
"""An array can repeat a name where a mapping cannot; which of two conflicting answers a
consumer acts on must not depend on server ordering."""
server = _Server(lambda request: httpx.Response(200, json={"answers": [_predicate_answer("q", 0.2), _predicate_answer("q", 0.9)]}))
answers = _ask(server, {"q": _PREDICATE})
assert answers.answers["q"] == NoulAnswer(probability=0.2)
def test_an_answer_for_a_question_that_was_never_asked_is_ignored_not_reported(self):
server = _Server(lambda request: httpx.Response(200, json={"answers": [_predicate_answer("q", 0.2), _predicate_answer("somebody_elses_question", 0.95)]}))
answers = _ask(server, {"q": _PREDICATE})
assert set(answers.answers) == {"q"}
assert answers.errors_by_question == {}
def test_a_missing_entry_is_reported_as_missing(self):
server = _Server(lambda request: httpx.Response(200, json={"answers": []}))
answers = _ask(server, {"q": _PREDICATE})
assert answers.errors_by_question["q"].category == CATEGORY_MISSING
class TestEnvelopeFailures:
def test_a_non_200_names_this_service_and_not_the_other_one(self):
server = _Server(lambda request: httpx.Response(500))
with pytest.raises(TypeSafeError) as excinfo:
_ask(server, {"q": _PREDICATE})
assert excinfo.value.cause == CAUSE_HTTP_STATUS
assert "OpenAI Decisions returned HTTP 500" in str(excinfo.value)
def test_an_answers_object_is_not_an_array_and_is_not_a_verdict(self):
server = _Server(lambda request: httpx.Response(200, json={"answers": {"q": {"probability": 0.9}}}))
with pytest.raises(TypeSafeError, match="'answers' array") as excinfo:
_ask(server, {"q": _PREDICATE})
assert excinfo.value.cause == CAUSE_INVALID_RESPONSE
def test_an_oversized_body_is_refused_rather_than_read(self):
filler = "x" * (MAX_RESPONSE_BYTES + 1024)
server = _Server(lambda request: httpx.Response(200, json={"answers": [_predicate_answer("q", 0.5), {"type": "refusal", "name": filler}]}))
with pytest.raises(TypeSafeError, match="exceeded") as excinfo:
_ask(server, {"q": _PREDICATE})
assert excinfo.value.cause == CAUSE_INVALID_RESPONSE
def test_a_compressed_body_is_refused_rather_than_expanded(self):
"""Refused by its encoding, not by a parse failure: this body decodes to a valid envelope."""
envelope = json.dumps({"answers": [_predicate_answer("q", 0.5)]}).encode()
server = _Server(lambda request: httpx.Response(200, content=gzip.compress(envelope), headers={"Content-Encoding": "gzip"}))
with pytest.raises(TypeSafeError, match="non-identity") as excinfo:
_ask(server, {"q": _PREDICATE})
assert excinfo.value.cause == CAUSE_INVALID_RESPONSE
# --- the tool gate ---------------------------------------------------------
def _gate(server: _Server, **kwargs) -> OpenAIDecisionsGuardrailProvider:
return OpenAIDecisionsGuardrailProvider(api_key=_API_KEY, model=_MODEL, transport_factory=server.transport, **kwargs)
def _tool_call() -> GuardrailRequest:
return GuardrailRequest(tool_name="bash", tool_input={"command": "ls"})
class TestGuardrailGate:
def test_a_probability_at_the_threshold_denies_and_names_this_provider(self):
server = _Server(lambda request: httpx.Response(200, json={"model": "served-1.2", "answers": [_predicate_answer("risky_tool_call", 0.5)]}))
decision = _gate(server, threshold=0.5).evaluate(_tool_call())
assert decision.allow is False
assert decision.reasons[0].code == "openai_decisions.tool_call_risky"
assert "model=served-1.2" in decision.reasons[0].message
def test_a_probability_below_the_threshold_allows(self):
server = _Server(lambda request: httpx.Response(200, json={"answers": [_predicate_answer("risky_tool_call", 0.4999)]}))
decision = _gate(server, threshold=0.5).evaluate(_tool_call())
assert decision.allow is True
assert decision.reasons[0].code == "openai_decisions.allowed"
def test_the_bytes_judged_are_the_bytes_the_audit_digest_identifies(self):
server = _Server(lambda request: httpx.Response(200, json={"answers": [_predicate_answer("risky_tool_call", 0.9)]}))
decision = _gate(server).evaluate(_tool_call())
sent = server.body["input"]
assert sent == _EXPECTED_GATE_INPUT
assert decision.metadata["state_digest"] == hashlib.sha256(sent.encode("utf-8")).hexdigest()[:16]
def test_a_refusal_denies_for_want_of_a_verdict(self):
"""The gate has no verdict without its answer, so a refusal is a denial here while the
same outcome is 'no opinion' for the memory sides."""
server = _Server(lambda request: httpx.Response(200, json={"answers": [{"type": "refusal", "name": "risky_tool_call"}]}))
with pytest.raises(OpenAIDecisionsGuardrailError) as excinfo:
_gate(server).evaluate(_tool_call())
assert excinfo.value.cause == CAUSE_INVALID_RESPONSE
assert "refused to answer" in str(excinfo.value)
def test_a_request_failure_is_the_provider_error_family_the_middleware_already_maps(self):
server = _Server(lambda request: httpx.Response(503))
with pytest.raises(OpenAIDecisionsGuardrailError):
_gate(server).evaluate(_tool_call())
def test_an_oversized_call_is_denied_locally_without_a_request(self):
server = _Server()
decision = _gate(server, max_state_chars=10).evaluate(GuardrailRequest(tool_name="bash", tool_input={"command": "rm -rf /tmp/anything-at-all"}))
assert decision.allow is False
assert decision.reasons[0].code == "openai_decisions.state_unusable"
assert server.requests == []
# --- the memory sides ------------------------------------------------------
def _prescreen(server: _Server, **kwargs) -> OpenAIDecisionsMemoryPrescreen:
return OpenAIDecisionsMemoryPrescreen(mode=MODE_ENFORCE, api_key=_API_KEY, model=_MODEL, transport_factory=server.transport, **kwargs)
class TestMemorySides:
def test_the_pre_screen_skips_below_its_threshold_on_the_same_text_it_always_judged(self):
server = _Server(lambda request: httpx.Response(200, json={"answers": [_predicate_answer(PRESCREEN_QUESTION, 0.1)]}))
provider = _prescreen(server, skip_threshold=0.2)
decision = provider.decide(_prescreen_request("user: keep using uv\nassistant: noted"))
assert decision.verdict == "skip"
assert server.body["input"] == "user: keep using uv\nassistant: noted"
assert server.body["questions"][0]["name"] == PRESCREEN_QUESTION
def test_a_refusal_is_no_opinion_so_the_batch_still_extracts(self):
server = _Server(lambda request: httpx.Response(200, json={"answers": [{"type": "refusal", "name": PRESCREEN_QUESTION}]}))
assert _prescreen(server).decide(_prescreen_request("anything")) is None
def test_both_directions_travel_in_one_request_and_a_refusal_only_costs_its_own_label(self):
server = _Server(
lambda request: httpx.Response(
200,
json={"answers": [_predicate_answer(QUESTION_AFFIRMATION, 0.8), {"type": "refusal", "name": QUESTION_NEGATION}]},
)
)
provider = OpenAIDecisionsSignalClassifier(mode="hints", api_key=_API_KEY, model=_MODEL, transport_factory=server.transport)
decision = provider.decide(_signal_request("user: yes keep using uv"))
assert [entry["name"] for entry in server.body["questions"]] == [QUESTION_AFFIRMATION, QUESTION_NEGATION]
assert decision.probabilities == {LABEL_REINFORCEMENT: 0.8}
assert decision.labels == frozenset({LABEL_REINFORCEMENT})
def test_the_two_backends_never_share_one_combined_request(self):
"""``sharing_key`` is what lets two sides travel together; a different endpoint or a
different credential must separate them or a verdict could be served to the wrong service."""
server = _Server()
settings = {"api_key": _API_KEY, "model": _MODEL, "base_url": "https://decision-gateway.example", "transport_factory": server.transport}
left = TypeSafeMemoryPrescreen(**settings)
right = OpenAIDecisionsMemoryPrescreen(**settings)
assert left._connection.url != right._connection.url
assert left.sharing_key() != right.sharing_key()
def test_it_resolves_from_the_existing_provider_slot_without_a_schema_change(self):
provider = resolve_memory_prescreen(
mode="shadow",
use="deerflow.agents.memory.prescreen.openai_decisions:OpenAIDecisionsMemoryPrescreen",
config={"api_key": _API_KEY, "model": _MODEL},
)
assert provider.name == "openai_decisions"
assert provider.policy_id == "deerflow.memory.prescreen.openai_decisions"
def _gateway_response(request: httpx.Request) -> httpx.Response:
"""One gateway serves both dialects, with different classifier opinions."""
questions = json.loads(request.content)["questions"]
if request.url.path == "/v1/systemone":
assert isinstance(questions, dict)
return httpx.Response(200, json={"model": "jev-test", "answers": {name: {"type": "noul", "noul": 0.1} for name in questions}})
assert request.url.path == "/v1/decisions"
probabilities = {PRESCREEN_QUESTION: 0.1, QUESTION_AFFIRMATION: 0.9, QUESTION_NEGATION: 0.1}
return httpx.Response(200, json={"answers": [_predicate_answer(question["name"], probabilities[question["name"]]) for question in questions]})
def _judge_gateway(prescreen_type, classifier_type, combine):
server = _Server(_gateway_response)
settings = {"api_key": _API_KEY, "model": _MODEL, "base_url": "https://decision-gateway.example", "transport_factory": server.transport}
coordinator = MemorySignalCoordinator(
prescreen=prescreen_type(mode="enforce", **settings),
prescreen_mode="enforce",
classifier=classifier_type(mode="hints", **settings),
classifier_mode="hints",
combine=combine,
)
text = "user: Continue using the same editor preference."
verdict = coordinator.judge(MemoryBatchContext(batch_text=text, digest=hashlib.sha256(text.encode()).hexdigest()[:16]))
return server, verdict
class TestSameGatewayCoordination:
@pytest.mark.parametrize("combine", ["auto", "never"])
@pytest.mark.parametrize(
"prescreen_type, classifier_type, paths, veto",
[
(TypeSafeMemoryPrescreen, OpenAIDecisionsSignalClassifier, ["/v1/systemone", "/v1/decisions"], True),
(OpenAIDecisionsMemoryPrescreen, TypeSafeSignalClassifier, ["/v1/decisions", "/v1/systemone"], False),
],
)
def test_mixed_backends_keep_their_requests_and_verdicts(self, combine, prescreen_type, classifier_type, paths, veto):
server, verdict = _judge_gateway(prescreen_type, classifier_type, combine)
assert [request.url.path for request in server.requests] == paths
for request, expected_questions in zip(server.requests, [{PRESCREEN_QUESTION}, {QUESTION_AFFIRMATION, QUESTION_NEGATION}], strict=True):
body = json.loads(request.content)
if request.url.path == "/v1/systemone":
assert set(body["questions"]) == expected_questions
assert "state" in body and "input" not in body
else:
assert {question["name"] for question in body["questions"]} == expected_questions
assert "input" in body and "state" not in body
assert verdict.skip is not veto
assert verdict.vetoed_by_model_signal is veto
assert verdict.hints == (frozenset({LABEL_REINFORCEMENT}) if veto else frozenset())
@pytest.mark.parametrize(
"prescreen_type, classifier_type, path, veto",
[
(TypeSafeMemoryPrescreen, TypeSafeSignalClassifier, "/v1/systemone", False),
(OpenAIDecisionsMemoryPrescreen, OpenAIDecisionsSignalClassifier, "/v1/decisions", True),
],
)
def test_matching_backends_still_combine(self, prescreen_type, classifier_type, path, veto):
server, verdict = _judge_gateway(prescreen_type, classifier_type, "auto")
assert [request.url.path for request in server.requests] == [path]
questions = json.loads(server.requests[0].content)["questions"]
names = set(questions) if path == "/v1/systemone" else {question["name"] for question in questions}
assert names == {PRESCREEN_QUESTION, QUESTION_AFFIRMATION, QUESTION_NEGATION}
assert verdict.skip is not veto
assert verdict.vetoed_by_model_signal is veto
def _prescreen_request(text: str):
from deerflow.agents.memory.prescreen.contract import MemoryPrescreenRequest
return MemoryPrescreenRequest(batch_text=text, digest=hashlib.sha256(text.encode()).hexdigest()[:16])
def _signal_request(text: str):
from deerflow.agents.memory.signals.contract import MemorySignalRequest
return MemorySignalRequest(batch_text=text, digest=hashlib.sha256(text.encode()).hexdigest()[:16])

View File

@ -192,6 +192,15 @@ class TestModeOff:
class TestIdentities:
def test_endpoint_path_separates_sharing_and_public_policy_identity(self):
settings = {"api_key": "key-a", "model": "gateway-model", "base_url": "https://gateway.example"}
jev = resolve_connection(settings=settings, configuration_source="tests.typesafe")
decisions = resolve_connection(settings=settings, configuration_source="tests.decisions", endpoint_path="/v1/decisions")
assert TypeSafeClient(jev).sharing_key() != TypeSafeClient(decisions).sharing_key()
assert jev.public_parameters()["endpoint_path"] == "/v1/systemone"
assert decisions.public_parameters()["endpoint_path"] == "/v1/decisions"
def test_the_sharing_key_tracks_every_sharing_dimension(self):
base = {"api_key": "key-a", "model": "jev-latest"}
baseline = _client(**base).sharing_key(max_state_chars=100)

View File

@ -3295,6 +3295,24 @@ run_ownership:
# # max_state_chars / cache_size / cache_ttl_seconds /
# # instructions / criteria
# --- Option 4b: the same risk gate over OpenAI Decisions -------------------
# Same provider semantics, another endpoint: one `predicate` question about the
# call to POST /v1/decisions, denied when its probability reaches `threshold`.
# Nothing in the gate is re-decided -- the strict-JSON local preflight, the
# character-count limit, the cache and fail_closed are the Option 4 ones, and
# every other Option 4 setting below applies here unchanged.
# guardrails:
# enabled: true
# fail_closed: true
# provider:
# use: deerflow.guardrails.openai_decisions:OpenAIDecisionsGuardrailProvider
# config:
# model: <name the decision model this deployment uses> # required: this
# # backend ships no default model, so batches cannot go to a model nobody chose
# api_key_env: OPENAI_API_KEY # default; base_url defaults to https://api.openai.com
# threshold: 0.5
# tools: ["bash"]
# ============================================================================
# TypeSafe (Jev) Client Defaults (optional)
# ============================================================================