feat: context-pressure handoff watchdog + multimodal RAG embedder protocol

[Gradata]

Summary

Two TikTok-derived features, both provider-agnostic and local-first.

• feat: context-pressure handoff watchdog (preemptive compaction) #127 — Context-pressure handoff watchdog (gradata.contrib.patterns.handoff). HandoffWatchdog.check(tokens_used, tokens_max) triggers a preemptive resume doc at 0.65 pressure (override via GRADATA_HANDOFF_THRESHOLD, clamped to [0.10, 0.95]). Fires once per lifecycle, writes a stable Markdown handoff, emits handoff.triggered event, reset() re-arms.
• feat: multimodal RAG for image/audio/video correction evidence #128 — Multimodal RAG embedder protocol (gradata.enhancements.rag.embedders). MultimodalEmbedder Protocol + MultimodalInput (text/image/audio/video) + TextOnlyEmbedder fallback + embed_any router. Users bring their own Gemini/Voyage/CLIP provider — Gradata never hosts the endpoint.

Test plan

• Unit tests: tests/test_handoff.py (18) + tests/test_rag_embedders.py (20)
• Full suite: 3853 passed, 3 skipped (fcntl platform gates — unfixable on Windows)
• Integration: wire HandoffWatchdog into brain session loop (follow-up PR)
• Integration: wire embed_any into similarity.py best_similarity path (follow-up PR)

Out of scope (deferred)

• Actual Gemini reference implementation (caller-supplied for now)
• SDK integration points for the watchdog beyond the module-level API
• Rule-schema upgrade (examples + output_shape) — tracked separately in feat: examples + output_shape on Lesson schema (prompt-injection standard) #129

Generated with Gradata

[greptile-apps]

Your free trial has ended. If you'd like to continue receiving code reviews, you can add a payment method here.

[coderabbitai]

📝 Walkthrough

• New features:
  • Context-pressure handoff watchdog (gradata.contrib.patterns.handoff): HandoffWatchdog (fires once per lifecycle; reset() to re-arm), HandoffDoc, measure_pressure(tokens_used, tokens_max), file helpers (load_handoff, pick_latest_unconsumed, consume_handoff), default threshold 0.65 via GRADATA_HANDOFF_THRESHOLD (clamped to [0.10,0.95]).
  • Multimodal RAG embedder protocol (gradata.enhancements.rag.embedders): MultimodalInput, MultimodalEmbedder Protocol (supports/embed), TextOnlyEmbedder fallback (blake2b deterministic text embeddings, L2-normalized), and embed_any(...) router for text/image/audio/video.
• New public APIs / exports:
  • gradata.contrib.patterns.handoff: measure_pressure, HandoffDoc, HandoffWatchdog, default_handoff_dir, pick_latest_unconsumed, consume_handoff, load_handoff.
  • gradata.enhancements.rag: MultimodalEmbedder, MultimodalInput, TextOnlyEmbedder, embed_any (re-exported in package init).
  • gradata.enhancements.rule_synthesizer.synthesize_rules_block(...).
  • New hook: gradata.hooks.inject_handoff.main(data) (injects latest unconsumed handoff at SessionStart).
• Behavioral / architectural shifts:
  • Meta-rule discovery and synthesis are now local-first (discover_meta_rules, merge_into_meta) with semantic clustering/decay; Cloud is reframed as visualization/team-sharing/backup and no longer gates or re-runs the learning loop (docs updated accordingly).
• Breaking / signature changes:
  • gradata._doctor.diagnose(...) signature changed to diagnose(..., include_cloud: bool = True, cloud_only: bool = False).
  • gradata.hooks.inject_brain_rules._lesson_to_rule_dict signature now accepts current_session: _lesson_to_rule_dict(lesson, current_session: int = 0).
  • CLI: gradata doctor gains --cloud / --no-cloud flags (wires into diagnose()).
• Tests & CI:
  • New/updated tests added (notably tests/test_handoff.py and tests/test_rag_embedders.py); full suite reported as 3853 passed, 3 skipped.
• Other notable changes:
  • Session hooks: SessionClose now attempts to resolve lesson_applications (PENDING → CONFIRMED/REJECTED) and refreshes an auto-generated brain_prompt.md via rule_synthesizer.
  • Cloud sync: Supabase env-var alias support, deterministic UUID upserts, table/row remapping and payload shaping in gradata._cloud_sync.
  • rule_synthesizer: fail-safe synthesis with Anthropic/CLI fallback and SHA256-based cache of extracted blocks.
• Security fixes: none indicated.

Walkthrough

Reframes Cloud as a read-only visualization/backup surface and moves graduation, meta-rule discovery/synthesis, and promotion to local execution; adds cloud-sync transformations and Supabase alias support, cloud diagnostics, a context-pressure handoff system and inject hook, RAG embedders, a rule synthesizer, many pipeline/hook changes and extensive tests/docs updates.

Changes

Cohort / File(s)	Summary
Docs: Cloud & Local-First Gradata/docs/architecture/cloud-monolith-v2.md, Gradata/docs/architecture/multi-tenant-future-proofing.md, Gradata/docs/cloud/overview.md, Gradata/docs/cloud/dashboard.md, Gradata/docs/concepts/meta-rules.md	Rewrote cloud responsibilities to read-only visualization/team-sharing/backup; clarified that graduation, meta-rule synthesis, and promotion run locally and cloud never re-runs graduation.
Docs: Legacy & Ignore Gradata/docs/LEGACY_CLEANUP.md, .gitignore	Added legacy-cleanup guidance and updated .gitignore to ignore Gradata/docs/pre-launch-plan.md.
Meta-rule system Gradata/src/gradata/enhancements/meta_rules.py	Replaced cloud-gated discovery with local-first clustering, noise filtering, decay/confidence adjustments, deterministic principle construction, and updated merge/synthesis semantics.
Cloud sync & client Gradata/src/gradata/_cloud_sync.py, Gradata/src/gradata/cloud/client.py	Added Supabase env-var alias resolution, table/row remapping (correction_patterns→corrections), deterministic per-row UUIDs, JSONB payload packing, batch dedupe/scrubbing, transformed push(), and changed cloud base URL to https://api.gradata.ai/api/v1.
Diagnostics / CLI Gradata/src/gradata/_doctor.py, Gradata/src/gradata/cli.py	Introduced cloud health checks (config/env/probes/auth/data), diagnose(..., include_cloud, cloud_only) signature, and CLI doctor flags --cloud/--no-cloud.
Handoff contrib & hook Gradata/src/gradata/contrib/patterns/handoff.py, Gradata/src/gradata/hooks/inject_handoff.py, Gradata/tests/test_handoff.py, Gradata/tests/test_inject_handoff_hook.py	New context-pressure helpers, HandoffDoc/HandoffWatchdog, file helpers, tests, and a hook to inject/latest-consumed handoff at SessionStart with consumption and event emission.
RAG embedder framework Gradata/src/gradata/enhancements/rag/embedders.py, Gradata/src/gradata/enhancements/rag/__init__.py, Gradata/tests/test_rag_embedders.py	Added multimodal input model, MultimodalEmbedder protocol, TextOnlyEmbedder, embed_any() routing/fallback and tests.
Rule synthesizer Gradata/src/gradata/enhancements/rule_synthesizer.py, Gradata/tests/test_rule_synthesizer.py	New synthesize_rules_block() with Anthropic/claude providers, strict prompt, SHA256-based cache, fail-safe fallbacks, and tests.
Graduation & pipeline Gradata/src/gradata/enhancements/graduation/agent_graduation.py, Gradata/src/gradata/enhancements/rule_pipeline.py	Formatting and logging changes; centralized graduation call usage, and adjusted confidence update behavior for rejections.
Hooks: inject_brain_rules & session_close Gradata/src/gradata/hooks/inject_brain_rules.py, Gradata/src/gradata/hooks/session_close.py	Persistent brain-prompt short-circuiting, _lesson_to_rule_dict signature change, DB writes of lesson_applications on injection, and post-stop pending-application resolution plus brain_prompt refresh via synthesizer.
Hooks: implicit feedback Gradata/src/gradata/hooks/implicit_feedback.py, Gradata/tests/test_implicit_feedback.py	Expanded regex coverage for shorthand/apostrophe-less negation, reminders, and broader challenge patterns; added tests.
Tests: additions & gating removal Gradata/tests/... (many files)	Added many new test modules (doctor cloud, handoff, rule_synthesizer, rag embedders, lesson_applications, etc.), removed cloud/LLM gating and several xfail/skips, updated expectations to local-first behavior, and adjusted various assertions.
Misc & formatting Gradata/tests/conftest.py, Gradata/src/gradata/enhancements/rag/..., others	Minor formatting/whitespace changes, new package exports, test blank-line adjustments, and other non-functional edits.

Sequence Diagram(s)

sequenceDiagram
    participant User as User/Client
    participant SDK as Local SDK
    participant Graduation as Graduation Engine
    participant Synth as Meta-Rule Synthesizer
    participant Inject as Injection Hooks
    participant CloudSync as Cloud Sync
    participant Cloud as Gradata Cloud

    User->>SDK: submit correction/event
    SDK->>SDK: persist event / create lesson
    SDK->>Graduation: run graduation pipeline
    Graduation->>SDK: return graduated lessons
    SDK->>Synth: discover/synthesize meta-rules (local)
    Synth->>SDK: return meta-rules/principles
    SDK->>Inject: promote rules / update brain_prompt / write audit rows
    SDK->>CloudSync: optional push()
    CloudSync->>Cloud: transform & upsert rows (outbound only)
    Cloud->>Cloud: render dashboard / backups (read-only)

Loading

Estimated code review effort

🎯 4 (Complex) | ⏱️ ~70 minutes

Possibly related issues

• feat: context-pressure handoff watchdog (preemptive compaction) #127 — Implements the context-pressure handoff/watchdog feature; this PR adds the same contrib.patterns.handoff module, tests, and inject hook.

Possibly related PRs

• feat: Sim 9 analytics — CUSUM, Beta scoring, kill threshold fix, dynamic similarity #16 — Strong overlap on meta-rule discovery/merge behavior changes; both modify gradata.enhancements.meta_rules.
• feat: S102 — MiroFish P0-P2 roadmap implementation (9 features) #24 — Related edits to implicit-feedback signal detection; both change gradata/hooks/implicit_feedback.py.
• perf(hooks): gate session_close + scrub loop-state + working session counter #106 — Related post-stop/session-close behavior; both touch gradata/hooks/session_close.py and its post-pipeline resolution/synthesis steps.

🚥 Pre-merge checks | ✅ 4 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 41.47% which is insufficient. The required threshold is 80.00%.	Write docstrings for the functions missing them to satisfy the coverage threshold.

✅ Passed checks (4 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title accurately describes the two main feature additions in the PR: context-pressure handoff watchdog and multimodal RAG embedder protocol.
Description check	✅ Passed	The description provides comprehensive context about the PR objectives, the two features added, test coverage, and deferred items.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing Touches

📝 Generate docstrings

• Create stacked PR
• Commit on current branch

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch feat/tiktok-handoff-and-multimodal-rag

---

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[greptile-apps]

Your free trial has ended. If you'd like to continue receiving code reviews, you can add a payment method here.

[coderabbitai]

Actionable comments posted: 22

Caution

Some comments are outside the diff and can’t be posted inline due to platform limitations.

⚠️ Outside diff range comments (4)

Gradata/tests/test_cloud_row_push.py (1)

15-20: ⚠️ Potential issue | 🟠 Major

Clear the new alias env vars in this fixture too.

_cloud_sync.enabled() now also reads GRADATA_SUPABASE_URL and GRADATA_SUPABASE_SERVICE_KEY, but this fixture only deletes GRADATA_CLOUD_*. test_disabled_without_credentials will become environment-dependent if those aliases are set in CI or a developer shell.

Proposed fix

     monkeypatch.delenv(_cloud_sync.ENV_ENABLED, raising=False)
     monkeypatch.delenv(_cloud_sync.ENV_URL, raising=False)
     monkeypatch.delenv(_cloud_sync.ENV_KEY, raising=False)
+    monkeypatch.delenv(_cloud_sync.ENV_URL_ALIAS, raising=False)
+    monkeypatch.delenv(_cloud_sync.ENV_KEY_ALIAS, raising=False)
     (tmp_path / ".tenant_id").write_text("11111111-2222-3333-4444-555555555555", encoding="utf-8")

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@Gradata/tests/test_cloud_row_push.py` around lines 15 - 20, The fixture
function brain currently clears only _cloud_sync.ENV_ENABLED,
_cloud_sync.ENV_URL, and _cloud_sync.ENV_KEY; update it to also delete the new
alias environment variables (GRADATA_SUPABASE_URL and
GRADATA_SUPABASE_SERVICE_KEY) so _cloud_sync.enabled() is deterministic in
tests—add monkeypatch.delenv("GRADATA_SUPABASE_URL", raising=False) and
monkeypatch.delenv("GRADATA_SUPABASE_SERVICE_KEY", raising=False) alongside the
existing deletions inside the brain fixture.

Gradata/src/gradata/hooks/inject_brain_rules.py (2)

441-455: ⚠️ Potential issue | 🟠 Major

lesson_applications and .last_injection.json no longer match the actual SessionStart payload.

This code persists audit state from the ranked cluster/member fragments, but if brain_prompt.md exists the hook returns bp_text instead. That makes correction attribution and PENDING/CONFIRMED resolution track a different prompt than the model actually received.

Suggested fix

+    bp_text = _read_brain_prompt(Path(brain_dir))
+    if bp_text:
+        return {"result": bp_text}
+
     lines = cluster_lines + individual_lines
     rules_block = "<brain-rules>\n" + "\n".join(lines) + "\n</brain-rules>"
@@
-    # Persistent brain-prompt: if brain/brain_prompt.md exists AND was written
-    # by session_close._refresh_brain_prompt (identified by the AUTO-GENERATED
-    # header), inject it verbatim and skip the fragmented composition.
-    # Synthesis never runs in the injection hook — that path was slow (CLI
-    # round-trip) and non-deterministic. The session_close hook is the only
-    # place we call the LLM; injection is pure read-compose.
-    bp_text = _read_brain_prompt(Path(brain_dir))
-    if bp_text:
-        return {"result": bp_text}
-
     return {"result": mandatory_block + disposition_block + rules_block + meta_block}

Also applies to: 457-495, 607-615

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@Gradata/src/gradata/hooks/inject_brain_rules.py` around lines 441 - 455, The
hook currently writes injection_manifest to .last_injection.json even when the
code later returns bp_text (from brain_prompt.md), causing lesson_applications
and persisted audit state to diverge from the actual SessionStart payload;
update the write logic in the inject_brain_rules path (references:
injection_manifest, manifest_path, bp_text, brain_prompt.md,
lesson_applications) to record the exact payload used for the session — if
bp_text is returned use that content (or a derived payload object matching the
SessionStart structure) instead of injection_manifest; apply the same change to
the other persistence sites mentioned (the blocks around lines 457-495 and
607-615) so the .last_injection.json always reflects the actual SessionStart
payload and correction attribution will match.

---

145-170: ⚠️ Potential issue | 🔴 Critical

Do not interpolate session context into bash -c.

On Windows, context is inserted directly into a Git Bash command string. Quotes or shell metacharacters in the session/task name can execute arbitrary shell fragments.

Suggested fix

     if sys.platform == "win32":
         git_bash = shutil.which("bash", path="C:/Program Files/Git/bin")
         if git_bash:
-            cmd = [git_bash, "-c", f'qmd search "{context}" -c brain -n 10']
+            cmd = [
+                git_bash,
+                "-lc",
+                'qmd search "$1" -c brain -n 10',
+                "bash",
+                context,
+            ]
         else:
             # Loud fallback: wiki-aware routing is silently disabled without

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@Gradata/src/gradata/hooks/inject_brain_rules.py` around lines 145 - 170, On
Windows the code builds a shell command string with the untrusted variable
context into cmd (the branch that finds git_bash) which allows shell injection;
change the Windows branch so context is safely quoted or, better, qmd is invoked
without embedding raw text into a "-c" shell string: import shlex and replace
the f-string with a safe quoted form using shlex.quote(context) when
constructing cmd (e.g. cmd = [git_bash, "-c", f'qmd search
{shlex.quote(context)} -c brain -n 10']) or, alternatively, avoid bash -c
entirely by invoking qmd directly as a list (cmd = ["qmd", "search", context,
"-c", "brain", "-n", "10"]) so subprocess.run (the call using cmd,
capture_output, text, timeout, encoding) is executed without interpolating raw
session context into a shell string; keep the _QMD_BASH_WARNED logic unchanged.

Gradata/src/gradata/enhancements/meta_rules.py (1)

345-396: ⚠️ Potential issue | 🟠 Major

Wire this local discovery path into refresh_meta_rules().

discover_meta_rules() is now local-first, but refresh_meta_rules() later in the same module still short-circuits with “Meta-rule discovery requires Gradata Cloud” and never calls it. Callers using the refresh API will continue to return only validated existing metas, so new local discoveries never surface.

♻️ Suggested wiring

 def refresh_meta_rules(
     lessons: list[Lesson],
     existing_metas: list[MetaRule],
     recent_corrections: list[dict] | None = None,
@@
 ) -> list[MetaRule]:
@@
-    _log.info("Meta-rule discovery requires Gradata Cloud")
     corrections = recent_corrections or []
@@
-    valid.sort(key=lambda m: m.confidence, reverse=True)
-    return valid
+    discovered = discover_meta_rules(
+        lessons,
+        min_group_size=min_group_size,
+        current_session=current_session,
+        **kwargs,
+    )
+
+    merged: dict[str, MetaRule] = {m.id: m for m in valid}
+    for meta in discovered:
+        merged.setdefault(meta.id, meta)
+
+    return sorted(merged.values(), key=lambda m: m.confidence, reverse=True)

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@Gradata/src/gradata/enhancements/meta_rules.py` around lines 345 - 396,
refresh_meta_rules currently short-circuits when Gradata Cloud is unavailable
and never invokes the new local-first discovery; update refresh_meta_rules to
call discover_meta_rules(lessons, min_group_size=..., current_session=...) as
the fallback (or merge its results with validated metas) when the cloud path
returns no metas or is disabled, ensure you pass the same current_session used
elsewhere, apply the existing _apply_decay/sorting behavior and logging (same as
in discover_meta_rules), and remove the early return that blocks local discovery
so local meta-rules surface via the refresh API.

🤖 Prompt for all review comments with AI agents

Verify each finding against the current code and only fix it if needed.

Inline comments:
In `@Gradata/docs/cloud/dashboard.md`:
- Line 3: Update the stale bullet in the Brain detail view that currently reads
“Meta-rules — cloud-synthesized principles” to reflect that meta-rule synthesis
runs locally in the SDK (e.g., change the phrase to “Meta-rules — locally
synthesized principles” or similar) so it matches the intro; locate the string
in the Gradata/docs/cloud/dashboard.md content under the Brain detail section
and replace the wording to indicate local synthesis.

In `@Gradata/src/gradata/_cloud_sync.py`:
- Around line 411-425: The current logic sets all_ok = False when
_transform_row(...) raises, which prevents _mark_push(conn, tenant_id, started)
from advancing the watermark; change the flow so malformed rows are
recorded/logged but do not flip the all_ok flag. Concretely, in the loop over
rows in the push routine keep a separate malformed_rows list or counter and in
the except block log and append to malformed_rows but do not set all_ok = False;
only set all_ok = False based on failures from _post(...) (e.g. when accepted !=
len(transformed) or _post indicates an error). Ensure pushed and _mark_push(...)
still run when there are accepted rows so the watermark advances even if some
rows were skipped by _transform_row; reference symbols: _transform_row, _post,
_mark_push, pushed, all_ok, transformed.
- Around line 140-147: The code currently coerces any numeric-looking
`session_raw` (e.g., "4.5") to an int which corrupts non-integral sessions;
change the logic around `session_raw`/`session_int` so you only set
`session_int` when `session_raw` is already an int or a string that represents
an integer (no decimal point or exponent), otherwise set `session_int = None`
and store the original value in `data_blob["session_raw"]` (if not already
present). Update the block handling `session_raw`, `session_int`, and
`data_blob` to perform that strict integer check instead of unconditionally
calling int().

In `@Gradata/src/gradata/_doctor.py`:
- Around line 305-320: The _check_cloud_reachable function currently extracts
the host via string splits and always connects to port 443; update it to parse
the api_url using urllib.parse.urlparse (use _read_cloud_config() result or
GRADATA_API_URL) to derive scheme, hostname and explicit port (falling back to
80 for http and 443 for https) and then call socket.create_connection((hostname,
port), timeout=_CLOUD_PROBE_TIMEOUT). Ensure the returned detail string includes
the actual host:port and preserve the existing return names/status semantics and
exception handling (refer to _check_cloud_reachable, _read_cloud_config, and
_CLOUD_PROBE_TIMEOUT).
- Around line 413-421: _diagnose currently invokes _cloud_checks
unconditionally, and _cloud_checks runs
_check_cloud_reachable/_check_cloud_auth/_check_cloud_has_data even when no
cloud config/env exists; change this so optional cloud probes don't fail default
doctor runs by gating them on actual cloud usage. Concretely: in
_cloud_checks(), call _check_cloud_config() (and/or implement a helper like
_cloud_enabled()) first and if it indicates cloud is not configured, either
return a list that only contains lightweight config/env checks or return stubbed
"skip"/"warn" check callables for _check_cloud_reachable, _check_cloud_auth, and
_check_cloud_has_data; alternatively modify those three check functions
themselves to detect "no config/no env" and immediately return a skip/warn
result rather than performing reachability/auth checks. Ensure diagnose() can
still include _cloud_checks() without risking false "broken" status when cloud
is unused.

In `@Gradata/src/gradata/cli.py`:
- Around line 271-273: Replace the stale fallback print message that mentions
the future cloud service with a local-first remediation telling users to install
or enable the reporting module; locate the print that currently outputs "Health
reports require the reporting module. Cloud features require the Gradata cloud
service (coming soon)." in gradata/cli.py (and the identical print around lines
306–308) and change it to a single-line message that omits any reference to the
cloud service and instead instructs users to install/enable the reporting
implementation (e.g., "Health reports require the reporting module; please
install or enable the reporting implementation.").
- Around line 224-231: The CLI is passing raw args.brain_dir into diagnose()
which ignores the project's env-first resolution (GRADATA_BRAIN > --brain-dir >
cwd); update the code so that brain_dir is the resolved path used elsewhere in
cli.py (i.e., use the same env-first resolver the CLI uses or compute brain_dir
by checking os.environ["GRADATA_BRAIN"] then args.brain_dir then cwd) and pass
that resolved brain_dir into diagnose(); ensure you reference the existing
brain_dir variable and the diagnose() call so the resolved value replaces
getattr(args, "brain_dir", None).
- Around line 1194-1195: The two flags for the doctor command (--cloud and
--no-cloud) must be made mutually exclusive so the CLI fails fast when both are
provided; locate the ArgumentParser/group setup where p_doctor is defined and
replace the separate add_argument calls for "--cloud" and "--no-cloud" with a
mutually exclusive group (use p_doctor.add_mutually_exclusive_group()) and add
both arguments to that group, keeping their action and help text identical so
the parser will error if both are passed.

In `@Gradata/src/gradata/cloud/client.py`:
- Around line 49-51: The endpoint resolution in the Gradata client
(self.endpoint in Gradata/src/gradata/cloud/client.py) currently only reads
ENV_ENDPOINT but must also honor the CLI/doctor environment variable
GRADATA_API_URL; update the lookup so it checks GRADATA_API_URL first (falling
back to ENV_ENDPOINT and then DEFAULT_ENDPOINT), strip any trailing "/" as
before, and preserve existing behavior when an explicit endpoint argument is
provided; reference the constants/identifiers ENV_ENDPOINT, DEFAULT_ENDPOINT and
the self.endpoint assignment to locate and change the code.

In `@Gradata/src/gradata/contrib/patterns/handoff.py`:
- Around line 133-136: The filename construction in _write uses raw doc.task_id
and doc.agent_name allowing path traversal or invalid characters; create and
call a sanitizer (e.g., sanitize_filename) that removes or replaces path
separators, '..', and platform-reserved characters (allow only a safe whitelist
like alphanumerics, dash, underscore) and use it in _write (and the
corresponding load/read method that builds the same
f"{task_id}_{agent_name}.handoff.md" at lines ~156-158) so all filenames are
safe and cannot escape handoff_dir.
- Around line 138-153: The telemetry emission in _emit should be best-effort:
wrap the call to events.emit inside a try/except that catches and suppresses any
exceptions (optionally logging them at debug) so emission failures don't
propagate; ensure that self._fired is set regardless of emit success (i.e.,
don't allow exceptions from events.emit to prevent flipping self._fired in the
calling flow). Update the _emit method to import and call events.emit inside a
safe try/except block and make sure the code that sets self._fired (and any
subsequent handoff state changes) cannot be skipped due to an emit exception.
- Around line 31-41: The _read_threshold function should clamp out-of-range or
invalid numeric overrides to the nearest bound instead of returning
_DEFAULT_THRESHOLD and must reject NaN; update _read_threshold to parse
GRADATA_HANDOFF_THRESHOLD to float, treat empty/missing as _DEFAULT_THRESHOLD,
return _MIN_THRESHOLD when value < _MIN_THRESHOLD, return _MAX_THRESHOLD when
value > _MAX_THRESHOLD, and if float(raw) raises or yields NaN (use math.isnan),
fall back to _DEFAULT_THRESHOLD; reference the _read_threshold function and the
constants _MIN_THRESHOLD, _MAX_THRESHOLD, and _DEFAULT_THRESHOLD when making
this change.

In `@Gradata/src/gradata/enhancements/graduation/agent_graduation.py`:
- Around line 219-223: The owner_only regex is using the literal token
"EXCLUDED_NAMES_PLACEHOLDER" so compile_deterministic_rule() compiles a
non-matching pattern and DATA_INTEGRITY silently passes; update the code that
builds/compiles rules (the compile_deterministic_rule() call) to replace/hydrate
EXCLUDED_NAMES_PLACEHOLDER with the actual excluded owner names from the brain
config (or return None from compile_deterministic_rule() if the brain config is
not yet available) so the owner_only rule becomes a real guard; ensure you
adjust the rule generation that references "owner_only" and the literal
"EXCLUDED_NAMES_PLACEHOLDER" accordingly and keep the compiled result None-safe
so DATA_INTEGRITY only registers when a valid regex is produced.

In `@Gradata/src/gradata/enhancements/meta_rules.py`:
- Around line 427-450: The code currently derives task_type from categories[0]
which is just the alphabetic first category; instead compute task_type from the
representative lesson (best.category) or the dominant category before building
applies_when/context_weights/scope. Replace uses of primary_cat = categories[0]
and any subsequent lookups (_CATEGORY_TASK_MAP.get(primary_cat,...)) with a
variable such as dominant_cat = best.category (or a function that returns the
dominant category) and use _CATEGORY_TASK_MAP.get(dominant_cat, "general") to
set task_type, then build applies_when, context_weights, and scope from that
task_type when constructing the MetaRule (affecting the block that creates
applies_when, context_weights, examples and scope).

In `@Gradata/src/gradata/enhancements/rag/embedders.py`:
- Around line 110-115: The code currently lets a NotImplementedError from
multimodal.embed propagate instead of falling back; update embed_any() so that
when multimodal is not None and multimodal.supports(item.modality) you call
multimodal.embed(item) inside a try/except that catches NotImplementedError and,
on that exception, proceeds to the existing text fallback logic (use
text_fallback or TextOnlyEmbedder()); keep using MultimodalEmbedder.supports()
to gate the call but handle runtime rejections by falling back to
text_fallback/TextOnlyEmbedder instead of propagating the error.

In `@Gradata/src/gradata/enhancements/rule_synthesizer.py`:
- Around line 195-198: The code returns whatever _read_cache(brain_dir,
cache_key) yields without validation; change the path in the cached-branch (the
block using cached, e.g., the cached variable and cache_key in
rule_synthesizer.py) to validate the cached entry shape and content before
returning: implement/inline checks for required fields (e.g., expected keys,
types, and schema/version), log a warning and ignore/delete the cache if it
fails validation, and return None so provider fallback runs; ensure any
exceptions during validation are caught, logged with cache_key, and treated as
"cache miss" rather than propagating malformed data.

In `@Gradata/src/gradata/hooks/session_close.py`:
- Around line 187-220: When no replacement rules block is generated (after
parsing lessons via parse_lessons and calling synthesize_rules_block using
lessons_path/bd), the code currently returns leaving the old AUTO-GENERATED
brain_prompt.md intact; update the early-return paths (the filtered-empty and
the block-empty cases shown and the similar case at 231-234) to delete or
overwrite the existing brain_prompt.md (or write a clear invalidation marker) so
inject_brain_rules._read_brain_prompt() will not pick up stale auto-generated
content on the next SessionStart.

In `@Gradata/tests/test_doctor_cloud.py`:
- Around line 130-139: The test test_diagnose_cloud_only calls
_doctor.diagnose(cloud_only=True) which triggers the real network probe
_check_cloud_reachable(); update the test to patch/mock the reachability probe
so no real socket is opened (e.g., monkeypatch or unittest.mock.patch of
_doctor._check_cloud_reachable or the underlying probe function) and make it
return a deterministic reachable/unreachable value; ensure the mock is applied
before calling _doctor.diagnose and restored after the test so names set
assertion remains unchanged.

In `@Gradata/tests/test_handoff.py`:
- Around line 39-45: Update the tests to match the documented clamp behavior of
_read_threshold: change the existing test_out_of_range_falls_back so that when
monkeypatch.setenv("GRADATA_HANDOFF_THRESHOLD","1.5") it asserts
_read_threshold() == 0.95 (high values clamp to 0.95); add/replace a test for
the lower bound by setting "GRADATA_HANDOFF_THRESHOLD" to "0.05" and asserting
_read_threshold() == 0.10 (low values clamp to 0.10); keep a separate test that
sets a non-numeric value (e.g. "not-a-number") and asserts the fallback default
(0.65) to preserve the garbage input behavior. Ensure references are to
_read_threshold and the test functions test_out_of_range_falls_back /
test_garbage_falls_back (or rename to reflect high/low/garbage cases).

In `@Gradata/tests/test_llm_synthesizer.py`:
- Around line 128-140: The test test_merge_produces_principle is too
weak—replace the loose assertion on meta.principle with a deterministic check
that the chosen principle matches the highest-priority lesson text passed to
merge_into_meta; specifically, after calling merge_into_meta(lessons, ...),
assert meta.principle == the exact string from the first lesson ("Use specific
infrastructure terms instead of follow-up phrasing") and keep the existing
meta.id assertion (meta.id.startswith("META-")) to ensure id format remains
validated.

In `@Gradata/tests/test_meta_rules.py`:
- Around line 85-86: The test currently recomputes the smoothing formula inline
which can hide regressions; replace the computed assertion using len(lessons)
with a concrete expected numeric value for this fixture (e.g. assert
meta.confidence == 0.5) so the test asserts an explicit expected confidence for
meta.confidence given the known lessons setup.

In `@Gradata/tests/test_rule_pipeline.py`:
- Around line 127-152: Replace fragile substring assertions on updated_text with
structured state checks: after calling run_rule_pipeline(lessons_path, db_path,
current_session=5), read and deserialize the lessons file into Lesson objects
(use the serializer/deserializer paired with _write_lessons or add a
_read_lessons helper) and assert the lesson's state equals LessonState.INSTINCT
and not LessonState.PATTERN. Locate the test
test_pipeline_does_not_graduate_at_exact_pattern_threshold, the helpers
_make_lesson and _write_lessons, and the pipeline invocation run_rule_pipeline
to implement the parsing and replace the assert "INSTINCT" / assert "PATTERN"
substring checks with explicit LessonState comparisons.

---

Outside diff comments:
In `@Gradata/src/gradata/enhancements/meta_rules.py`:
- Around line 345-396: refresh_meta_rules currently short-circuits when Gradata
Cloud is unavailable and never invokes the new local-first discovery; update
refresh_meta_rules to call discover_meta_rules(lessons, min_group_size=...,
current_session=...) as the fallback (or merge its results with validated metas)
when the cloud path returns no metas or is disabled, ensure you pass the same
current_session used elsewhere, apply the existing _apply_decay/sorting behavior
and logging (same as in discover_meta_rules), and remove the early return that
blocks local discovery so local meta-rules surface via the refresh API.

In `@Gradata/src/gradata/hooks/inject_brain_rules.py`:
- Around line 441-455: The hook currently writes injection_manifest to
.last_injection.json even when the code later returns bp_text (from
brain_prompt.md), causing lesson_applications and persisted audit state to
diverge from the actual SessionStart payload; update the write logic in the
inject_brain_rules path (references: injection_manifest, manifest_path, bp_text,
brain_prompt.md, lesson_applications) to record the exact payload used for the
session — if bp_text is returned use that content (or a derived payload object
matching the SessionStart structure) instead of injection_manifest; apply the
same change to the other persistence sites mentioned (the blocks around lines
457-495 and 607-615) so the .last_injection.json always reflects the actual
SessionStart payload and correction attribution will match.
- Around line 145-170: On Windows the code builds a shell command string with
the untrusted variable context into cmd (the branch that finds git_bash) which
allows shell injection; change the Windows branch so context is safely quoted
or, better, qmd is invoked without embedding raw text into a "-c" shell string:
import shlex and replace the f-string with a safe quoted form using
shlex.quote(context) when constructing cmd (e.g. cmd = [git_bash, "-c", f'qmd
search {shlex.quote(context)} -c brain -n 10']) or, alternatively, avoid bash -c
entirely by invoking qmd directly as a list (cmd = ["qmd", "search", context,
"-c", "brain", "-n", "10"]) so subprocess.run (the call using cmd,
capture_output, text, timeout, encoding) is executed without interpolating raw
session context into a shell string; keep the _QMD_BASH_WARNED logic unchanged.

In `@Gradata/tests/test_cloud_row_push.py`:
- Around line 15-20: The fixture function brain currently clears only
_cloud_sync.ENV_ENABLED, _cloud_sync.ENV_URL, and _cloud_sync.ENV_KEY; update it
to also delete the new alias environment variables (GRADATA_SUPABASE_URL and
GRADATA_SUPABASE_SERVICE_KEY) so _cloud_sync.enabled() is deterministic in
tests—add monkeypatch.delenv("GRADATA_SUPABASE_URL", raising=False) and
monkeypatch.delenv("GRADATA_SUPABASE_SERVICE_KEY", raising=False) alongside the
existing deletions inside the brain fixture.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: ASSERTIVE

Plan: Pro

Run ID: 30e1e68b-9d70-4f13-8546-9efaf25fa564

📥 Commits

Reviewing files that changed from the base of the PR and between 5ef1a1e and 3698836.

📒 Files selected for processing (39)

• .gitignore
• Gradata/docs/LEGACY_CLEANUP.md
• Gradata/docs/architecture/cloud-monolith-v2.md
• Gradata/docs/architecture/multi-tenant-future-proofing.md
• Gradata/docs/cloud/dashboard.md
• Gradata/docs/cloud/overview.md
• Gradata/docs/concepts/meta-rules.md
• Gradata/src/gradata/_cloud_sync.py
• Gradata/src/gradata/_doctor.py
• Gradata/src/gradata/cli.py
• Gradata/src/gradata/cloud/client.py
• Gradata/src/gradata/contrib/patterns/handoff.py
• Gradata/src/gradata/enhancements/graduation/agent_graduation.py
• Gradata/src/gradata/enhancements/meta_rules.py
• Gradata/src/gradata/enhancements/rag/__init__.py
• Gradata/src/gradata/enhancements/rag/embedders.py
• Gradata/src/gradata/enhancements/rule_pipeline.py
• Gradata/src/gradata/enhancements/rule_synthesizer.py
• Gradata/src/gradata/hooks/implicit_feedback.py
• Gradata/src/gradata/hooks/inject_brain_rules.py
• Gradata/src/gradata/hooks/session_close.py
• Gradata/tests/conftest.py
• Gradata/tests/test_agent_graduation.py
• Gradata/tests/test_bug_fixes.py
• Gradata/tests/test_cloud_row_push.py
• Gradata/tests/test_doctor_cloud.py
• Gradata/tests/test_handoff.py
• Gradata/tests/test_implicit_feedback.py
• Gradata/tests/test_integration_workflow.py
• Gradata/tests/test_lesson_applications.py
• Gradata/tests/test_llm_synthesizer.py
• Gradata/tests/test_mem0_adapter.py
• Gradata/tests/test_meta_rule_generalization.py
• Gradata/tests/test_meta_rules.py
• Gradata/tests/test_multi_brain_simulation.py
• Gradata/tests/test_pipeline_e2e.py
• Gradata/tests/test_rag_embedders.py
• Gradata/tests/test_rule_pipeline.py
• Gradata/tests/test_rule_synthesizer.py

💤 Files with no reviewable changes (2)

• Gradata/tests/test_bug_fixes.py
• Gradata/tests/test_multi_brain_simulation.py

📜 Review details

🧰 Additional context used

🧠 Learnings (2)

📓 Common learnings

Learnt from: Gradata
Repo: Gradata/gradata PR: 0
File: :0-0
Timestamp: 2026-04-17T17:18:07.439Z
Learning: In PR `#102` (gradata/gradata), Round 2 addressed: cli.py env-first brain resolution (GRADATA_BRAIN > --brain-dir > cwd), _tenant.py corrupt .tenant_id overwrite, _env_int default clamping to minimum, and _events.py tenant-scoped fallback SELECT for dedup. All ruff and 99 tests green after these fixes.

📚 Learning: 2026-04-17T17:18:07.439Z

Learnt from: Gradata
Repo: Gradata/gradata PR: 0
File: :0-0
Timestamp: 2026-04-17T17:18:07.439Z
Learning: In PR `#102` (gradata/gradata), Round 2 addressed: cli.py env-first brain resolution (GRADATA_BRAIN > --brain-dir > cwd), _tenant.py corrupt .tenant_id overwrite, _env_int default clamping to minimum, and _events.py tenant-scoped fallback SELECT for dedup. All ruff and 99 tests green after these fixes.

Applied to files:

• Gradata/tests/conftest.py
• Gradata/tests/test_cloud_row_push.py
• Gradata/tests/test_integration_workflow.py
• Gradata/src/gradata/hooks/session_close.py
• Gradata/src/gradata/_cloud_sync.py
• Gradata/tests/test_lesson_applications.py
• Gradata/src/gradata/_doctor.py
• Gradata/src/gradata/cli.py
• Gradata/docs/architecture/multi-tenant-future-proofing.md
• Gradata/tests/test_doctor_cloud.py
• Gradata/tests/test_pipeline_e2e.py

🪛 LanguageTool

Gradata/docs/cloud/overview.md

[style] ~3-~3: ‘on top of that’ might be wordy. Consider a shorter alternative.
Context: ...uity, team sharing, and managed backups on top of that local loop. ## What's in the SDK vs th...

(EN_WORDINESS_PREMIUM_ON_TOP_OF_THAT)

🪛 Ruff (0.15.10)

Gradata/src/gradata/_doctor.py

[warning] 337-340: Use contextlib.suppress(Exception) instead of try-except-pass

Replace try-except-pass with with contextlib.suppress(Exception): ...

(SIM105)

🔇 Additional comments (8)

.gitignore (1)

138-138: Good targeted ignore rule for private launch draft.

This is consistent with the surrounding Gradata/docs/* private-draft exclusions and helps prevent accidental publication.

Gradata/src/gradata/cloud/client.py (1)

68-75: Nice refactor: _post(path, payload) calls are cleaner and behavior-preserving.

Line 68-75 and Line 132-138 improve readability without changing payload semantics.

Also applies to: 132-138

Gradata/src/gradata/enhancements/rag/__init__.py (1)

3-15: Clean public API re-export for RAG embedders.

Good package boundary: explicit imports + __all__ keep the module surface predictable.

Gradata/docs/architecture/cloud-monolith-v2.md (1)

8-12: Local-first/cloud-reflection positioning is clearly stated.

This wording removes ambiguity about where learning logic executes and what cloud is responsible for.

Gradata/tests/test_llm_synthesizer.py (1)

47-119: Mocked LLM-path test cleanup looks good.

The refactor preserves behavior while keeping test setup/readability tighter.

Gradata/docs/concepts/meta-rules.md (1)

47-50: Excellent clarification of local synthesis vs cloud responsibilities.

This update makes execution boundaries explicit and reduces operator confusion.

Gradata/tests/test_meta_rule_generalization.py (1)

66-83: Same-category discovery test data is much clearer now.

The updated phrasing makes the expected thematic merge easier to understand and maintain.

Gradata/tests/test_agent_graduation.py (1)

396-486: Great regression coverage for fire-count category gating (H2).

These tests directly protect against silent over-promotion and match the intended evidence model.

[coderabbitai]

⚠️ Potential issue | 🟡 Minor

One bullet below is still stale.

This intro now says meta-rule synthesis runs locally in the SDK, but the Brain detail view still says “Meta-rules — cloud-synthesized principles”. Please update that bullet too so the page is internally consistent.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@Gradata/docs/cloud/dashboard.md` at line 3, Update the stale bullet in the
Brain detail view that currently reads “Meta-rules — cloud-synthesized
principles” to reflect that meta-rule synthesis runs locally in the SDK (e.g.,
change the phrase to “Meta-rules — locally synthesized principles” or similar)
so it matches the intro; locate the string in the
Gradata/docs/cloud/dashboard.md content under the Brain detail section and
replace the wording to indicate local synthesis.

[coderabbitai]

⚠️ Potential issue | 🟠 Major

Don't truncate non-integral session values.

Line 143 silently turns values like 4.5 into 4. That contradicts the comment above and corrupts cloud-side session analytics; non-integral values should fall back to data["session_raw"] instead of being coerced.

Proposed fix

-        session_raw = row.get("session")
-        session_int: int | None
-        try:
-            session_int = int(session_raw) if session_raw is not None else None
-        except (ValueError, TypeError):
-            session_int = None
-            if "session_raw" not in data_blob:
-                data_blob["session_raw"] = session_raw
+        session_raw = row.get("session")
+        session_int: int | None = None
+        if isinstance(session_raw, int):
+            session_int = session_raw
+        elif isinstance(session_raw, float):
+            if session_raw.is_integer():
+                session_int = int(session_raw)
+            else:
+                data_blob.setdefault("session_raw", session_raw)
+        elif session_raw is not None:
+            try:
+                session_int = int(session_raw)
+            except (ValueError, TypeError):
+                data_blob.setdefault("session_raw", session_raw)

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@Gradata/src/gradata/_cloud_sync.py` around lines 140 - 147, The code
currently coerces any numeric-looking `session_raw` (e.g., "4.5") to an int
which corrupts non-integral sessions; change the logic around
`session_raw`/`session_int` so you only set `session_int` when `session_raw` is
already an int or a string that represents an integer (no decimal point or
exponent), otherwise set `session_int = None` and store the original value in
`data_blob["session_raw"]` (if not already present). Update the block handling
`session_raw`, `session_int`, and `data_blob` to perform that strict integer
check instead of unconditionally calling int().

[coderabbitai]

⚠️ Potential issue | 🟠 Major

Skipped malformed rows currently block watermark advancement.

When _transform_row() fails, Line 417 forces all_ok = False, so Lines 424-425 never advance last_push_at. One bad historical row will therefore cause every later push() to resend the already accepted rows forever and prevent sync progress for that tenant. If malformed rows are intentionally skipped, they need a dead-letter path or they must not block the watermark.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@Gradata/src/gradata/_cloud_sync.py` around lines 411 - 425, The current logic
sets all_ok = False when _transform_row(...) raises, which prevents
_mark_push(conn, tenant_id, started) from advancing the watermark; change the
flow so malformed rows are recorded/logged but do not flip the all_ok flag.
Concretely, in the loop over rows in the push routine keep a separate
malformed_rows list or counter and in the except block log and append to
malformed_rows but do not set all_ok = False; only set all_ok = False based on
failures from _post(...) (e.g. when accepted != len(transformed) or _post
indicates an error). Ensure pushed and _mark_push(...) still run when there are
accepted rows so the watermark advances even if some rows were skipped by
_transform_row; reference symbols: _transform_row, _post, _mark_push, pushed,
all_ok, transformed.

[coderabbitai]

⚠️ Potential issue | 🟠 Major

Use the configured scheme and port in cloud_reachable.

Line 311 strips the URL with string splits and Line 313 always dials :443. That makes http://localhost:3000/... and https://host:8443/... report unreachable even when the API is healthy. Parse the URL and connect to the resolved hostname plus its explicit/default port instead.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@Gradata/src/gradata/_doctor.py` around lines 305 - 320, The
_check_cloud_reachable function currently extracts the host via string splits
and always connects to port 443; update it to parse the api_url using
urllib.parse.urlparse (use _read_cloud_config() result or GRADATA_API_URL) to
derive scheme, hostname and explicit port (falling back to 80 for http and 443
for https) and then call socket.create_connection((hostname, port),
timeout=_CLOUD_PROBE_TIMEOUT). Ensure the returned detail string includes the
actual host:port and preserve the existing return names/status semantics and
exception handling (refer to _check_cloud_reachable, _read_cloud_config, and
_CLOUD_PROBE_TIMEOUT).

[coderabbitai]

⚠️ Potential issue | 🟠 Major

Don't make an optional cloud probe break the default doctor run.

diagnose() includes _cloud_checks() by default, and _cloud_checks() always runs cloud_reachable even when no cloud config or env is present. On an offline machine, gradata doctor can become broken purely because the optional dashboard is unreachable, which hides the actual local health signal. Gate these probes on configured cloud usage, or downgrade them to skip/warn when Cloud is not in use.

Also applies to: 424-463

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@Gradata/src/gradata/_doctor.py` around lines 413 - 421, _diagnose currently
invokes _cloud_checks unconditionally, and _cloud_checks runs
_check_cloud_reachable/_check_cloud_auth/_check_cloud_has_data even when no
cloud config/env exists; change this so optional cloud probes don't fail default
doctor runs by gating them on actual cloud usage. Concretely: in
_cloud_checks(), call _check_cloud_config() (and/or implement a helper like
_cloud_enabled()) first and if it indicates cloud is not configured, either
return a list that only contains lightweight config/env checks or return stubbed
"skip"/"warn" check callables for _check_cloud_reachable, _check_cloud_auth, and
_check_cloud_has_data; alternatively modify those three check functions
themselves to detect "no config/no env" and immediately return a skip/warn
result rather than performing reachability/auth checks. Ensure diagnose() can
still include _cloud_checks() without risking false "broken" status when cloud
is unused.

[coderabbitai]

⚠️ Potential issue | 🟠 Major

Mock the reachability probe in the cloud-only test.

diagnose(cloud_only=True) still calls _check_cloud_reachable(), so this test opens a real socket unless you patch it. That makes an “offline, no real network calls” test flaky in no-egress CI.

Suggested test hardening

 def test_diagnose_cloud_only(isolated_config):
-    report = _doctor.diagnose(cloud_only=True)
+    with patch.object(
+        _doctor,
+        "_check_cloud_reachable",
+        return_value={"name": "cloud_reachable", "status": "skip", "detail": "mocked"},
+    ):
+        report = _doctor.diagnose(cloud_only=True)
     names = {c["name"] for c in report["checks"]}
     assert names == {
         "cloud_config",
         "cloud_env",
         "cloud_reachable",

📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change

	def test_diagnose_cloud_only(isolated_config):
	report = _doctor.diagnose(cloud_only=True)
	names = {c["name"] for c in report["checks"]}
	assert names == {
	"cloud_config",
	"cloud_env",
	"cloud_reachable",
	"cloud_auth",
	"cloud_has_data",
	}
	def test_diagnose_cloud_only(isolated_config):
	with patch.object(
	_doctor,
	"_check_cloud_reachable",
	return_value={"name": "cloud_reachable", "status": "skip", "detail": "mocked"},
	):
	report = _doctor.diagnose(cloud_only=True)
	names = {c["name"] for c in report["checks"]}
	assert names == {
	"cloud_config",
	"cloud_env",
	"cloud_reachable",
	"cloud_auth",
	"cloud_has_data",
	}

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@Gradata/tests/test_doctor_cloud.py` around lines 130 - 139, The test
test_diagnose_cloud_only calls _doctor.diagnose(cloud_only=True) which triggers
the real network probe _check_cloud_reachable(); update the test to patch/mock
the reachability probe so no real socket is opened (e.g., monkeypatch or
unittest.mock.patch of _doctor._check_cloud_reachable or the underlying probe
function) and make it return a deterministic reachable/unreachable value; ensure
the mock is applied before calling _doctor.diagnose and restored after the test
so names set assertion remains unchanged.

[coderabbitai]

⚠️ Potential issue | 🟠 Major

These assertions lock in the wrong threshold contract.

The feature is documented as clamping to [0.10, 0.95], not falling back to the default. Keeping 1.5 -> 0.65 here will fail the correct implementation and still misses the lower-bound case.

Suggested fix

-    def test_out_of_range_falls_back(self, monkeypatch):
+    def test_out_of_range_clamps(self, monkeypatch):
         monkeypatch.setenv("GRADATA_HANDOFF_THRESHOLD", "1.5")
-        assert _read_threshold() == 0.65
+        assert _read_threshold() == 0.95
+
+    def test_below_min_clamps(self, monkeypatch):
+        monkeypatch.setenv("GRADATA_HANDOFF_THRESHOLD", "0.01")
+        assert _read_threshold() == 0.10

📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change

	def test_out_of_range_falls_back(self, monkeypatch):
	monkeypatch.setenv("GRADATA_HANDOFF_THRESHOLD", "1.5")
	assert _read_threshold() == 0.65
	def test_garbage_falls_back(self, monkeypatch):
	monkeypatch.setenv("GRADATA_HANDOFF_THRESHOLD", "not-a-number")
	assert _read_threshold() == 0.65
	def test_out_of_range_clamps(self, monkeypatch):
	monkeypatch.setenv("GRADATA_HANDOFF_THRESHOLD", "1.5")
	assert _read_threshold() == 0.95
	def test_below_min_clamps(self, monkeypatch):
	monkeypatch.setenv("GRADATA_HANDOFF_THRESHOLD", "0.01")
	assert _read_threshold() == 0.10
	def test_garbage_falls_back(self, monkeypatch):
	monkeypatch.setenv("GRADATA_HANDOFF_THRESHOLD", "not-a-number")
	assert _read_threshold() == 0.65

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@Gradata/tests/test_handoff.py` around lines 39 - 45, Update the tests to
match the documented clamp behavior of _read_threshold: change the existing
test_out_of_range_falls_back so that when
monkeypatch.setenv("GRADATA_HANDOFF_THRESHOLD","1.5") it asserts
_read_threshold() == 0.95 (high values clamp to 0.95); add/replace a test for
the lower bound by setting "GRADATA_HANDOFF_THRESHOLD" to "0.05" and asserting
_read_threshold() == 0.10 (low values clamp to 0.10); keep a separate test that
sets a non-numeric value (e.g. "not-a-number") and asserts the fallback default
(0.65) to preserve the garbage input behavior. Ensure references are to
_read_threshold and the test functions test_out_of_range_falls_back /
test_garbage_falls_back (or rename to reflect high/low/garbage cases).

[coderabbitai]

🧹 Nitpick | 🔵 Trivial

Strengthen deterministic assertion beyond non-empty principle.

Current assertions can pass even if principle selection regresses. Consider asserting that the produced principle reflects the intended highest-priority lesson text.

♻️ Suggested test hardening

 def test_merge_produces_principle(self):
     from gradata.enhancements.meta_rules import merge_into_meta

     lessons = [
         _make_lesson(
             "Use specific infrastructure terms instead of follow-up phrasing", "CONTENT"
         ),
         _make_lesson("Replace hedging with concrete modernization language", "CONTENT"),
         _make_lesson("Swap vague openers for precise technical references", "CONTENT"),
     ]
+    lessons[0].confidence = 0.95  # make selection unambiguous
     meta = merge_into_meta(lessons, theme_override="content", session=1)
     assert meta.principle
+    assert "infrastructure" in meta.principle.lower()
     assert meta.id.startswith("META-")

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@Gradata/tests/test_llm_synthesizer.py` around lines 128 - 140, The test
test_merge_produces_principle is too weak—replace the loose assertion on
meta.principle with a deterministic check that the chosen principle matches the
highest-priority lesson text passed to merge_into_meta; specifically, after
calling merge_into_meta(lessons, ...), assert meta.principle == the exact string
from the first lesson ("Use specific infrastructure terms instead of follow-up
phrasing") and keep the existing meta.id assertion (meta.id.startswith("META-"))
to ensure id format remains validated.

[coderabbitai]

🧹 Nitpick | 🔵 Trivial

Use a concrete expected confidence value in this fixture.

Recomputing the formula in-test can mask regressions; fixed test data here allows asserting the explicit expected value.

♻️ Suggested assertion change

-    # Confidence uses count / (count + 3) smoothing (3 lessons → 0.50).
-    assert meta.confidence == round(len(lessons) / (len(lessons) + 3.0), 2)
+    # 3 lessons -> 3/(3+3)=0.50
+    assert meta.confidence == 0.50

📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change

	# Confidence uses count / (count + 3) smoothing (3 lessons → 0.50).
	assert meta.confidence == round(len(lessons) / (len(lessons) + 3.0), 2)
	# 3 lessons -> 3/(3+3)=0.50
	assert meta.confidence == 0.50

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@Gradata/tests/test_meta_rules.py` around lines 85 - 86, The test currently
recomputes the smoothing formula inline which can hide regressions; replace the
computed assertion using len(lessons) with a concrete expected numeric value for
this fixture (e.g. assert meta.confidence == 0.5) so the test asserts an
explicit expected confidence for meta.confidence given the known lessons setup.

[coderabbitai]

🧹 Nitpick | 🔵 Trivial

Prefer structured state assertions over substring checks.

These tests validate the right boundary behavior, but string containment can become flaky if serialization format changes.

♻️ Suggested robustness improvement

-    updated_text = lessons_path.read_text(encoding="utf-8")
-    assert "PATTERN" in updated_text
+    from gradata.enhancements.self_improvement import parse_lessons
+    updated = parse_lessons(lessons_path.read_text(encoding="utf-8"))
+    assert len(updated) == 1
+    assert updated[0].state == LessonState.PATTERN
...
-    updated_text = lessons_path.read_text(encoding="utf-8")
-    assert "INSTINCT" in updated_text
-    assert "PATTERN" not in updated_text
+    updated = parse_lessons(lessons_path.read_text(encoding="utf-8"))
+    assert len(updated) == 1
+    assert updated[0].state == LessonState.INSTINCT

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@Gradata/tests/test_rule_pipeline.py` around lines 127 - 152, Replace fragile
substring assertions on updated_text with structured state checks: after calling
run_rule_pipeline(lessons_path, db_path, current_session=5), read and
deserialize the lessons file into Lesson objects (use the
serializer/deserializer paired with _write_lessons or add a _read_lessons
helper) and assert the lesson's state equals LessonState.INSTINCT and not
LessonState.PATTERN. Locate the test
test_pipeline_does_not_graduate_at_exact_pattern_threshold, the helpers
_make_lesson and _write_lessons, and the pipeline invocation run_rule_pipeline
to implement the parsing and replace the assert "INSTINCT" / assert "PATTERN"
substring checks with explicit LessonState comparisons.

[coderabbitai]

♻️ Duplicate comments (5)

Gradata/src/gradata/enhancements/rag/embedders.py (1)

110-115: ⚠️ Potential issue | 🟠 Major

Catch provider NotImplementedError and continue fallback routing

At Line 110-115, embed_any() directly returns multimodal.embed(item). If a provider reports support but rejects a specific item at runtime via NotImplementedError, fallback never executes, which conflicts with the protocol/docstring contract.

Proposed fix

 def embed_any(
     item: MultimodalInput,
     *,
     multimodal: MultimodalEmbedder | None = None,
     text_fallback: MultimodalEmbedder | None = None,
 ) -> list[float]:
@@
     """
     if multimodal is not None and multimodal.supports(item.modality):
-        return multimodal.embed(item)
+        try:
+            return multimodal.embed(item)
+        except NotImplementedError:
+            # Provider declined this specific item at runtime; continue fallback path.
+            pass
 
     if item.modality == "text":
         embedder = text_fallback or TextOnlyEmbedder()
         return embedder.embed(item)

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@Gradata/src/gradata/enhancements/rag/embedders.py` around lines 110 - 115, In
embed_any(), when calling multimodal.embed(item) (after checking
multimodal.supports(item.modality)), wrap that call in a try/except that catches
NotImplementedError and allows execution to continue to the fallback routing
instead of returning immediately; specifically, catch NotImplementedError from
multimodal.embed(item), ignore or log it as appropriate, and then fall through
to the existing logic that selects text_fallback or TextOnlyEmbedder for
item.modality == "text" so runtime rejections by a provider trigger the
documented fallback behavior.

Gradata/src/gradata/contrib/patterns/handoff.py (3)

31-41: ⚠️ Potential issue | 🟠 Major

Threshold override contract is implemented incorrectly (fallback vs clamp).

At Line 39-Line 41, out-of-range numeric overrides currently fall back to 0.65, but the feature contract says they must clamp to [0.10, 0.95]. This also leaves non-finite floats unguarded.

Proposed fix

 import os
+import math
@@
 def _read_threshold() -> float:
@@
     try:
         value = float(raw)
     except ValueError:
         return _DEFAULT_THRESHOLD
-    if value < _MIN_THRESHOLD or value > _MAX_THRESHOLD:
+    if not math.isfinite(value):
         return _DEFAULT_THRESHOLD
-    return value
+    return max(_MIN_THRESHOLD, min(_MAX_THRESHOLD, value))

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@Gradata/src/gradata/contrib/patterns/handoff.py` around lines 31 - 41, The
_read_threshold function currently returns _DEFAULT_THRESHOLD when an override
is out-of-range or non-numeric; instead clamp numeric overrides into the allowed
range and reject only invalid/non-finite values; update _read_threshold to parse
the env var, ensure the parsed float is finite (not NaN/Inf), then return
min(max(value, _MIN_THRESHOLD), _MAX_THRESHOLD) for valid numbers and only
return _DEFAULT_THRESHOLD for parse failures or non-finite values (use the
existing symbols _DEFAULT_THRESHOLD, _MIN_THRESHOLD, _MAX_THRESHOLD and the
function _read_threshold to locate the change).

---

133-136: ⚠️ Potential issue | 🔴 Critical

Unsanitized filename components allow path traversal/invalid paths.

At Line 135 and Line 158, task_id/agent_name flow directly into path construction. This can escape handoff_dir or generate unsafe filenames.

Proposed fix

+import re
@@
+def _safe_name(value: str) -> str:
+    cleaned = re.sub(r"[^A-Za-z0-9._-]+", "_", value).strip("._")
+    return cleaned or "handoff"
+
@@
     def _write(self, doc: HandoffDoc) -> None:
         self.handoff_dir.mkdir(parents=True, exist_ok=True)
-        path = self.handoff_dir / f"{doc.task_id}_{doc.agent_name}.handoff.md"
+        path = self.handoff_dir / (
+            f"{_safe_name(doc.task_id)}_{_safe_name(doc.agent_name)}.handoff.md"
+        )
         path.write_text(doc.render(), encoding="utf-8")
@@
 def load_handoff(task_id: str, agent_name: str, handoff_dir: Path) -> str | None:
@@
-    path = Path(handoff_dir) / f"{task_id}_{agent_name}.handoff.md"
+    path = Path(handoff_dir) / (
+        f"{_safe_name(task_id)}_{_safe_name(agent_name)}.handoff.md"
+    )

Also applies to: 156-158

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@Gradata/src/gradata/contrib/patterns/handoff.py` around lines 133 - 136, The
filename is built directly from doc.task_id and doc.agent_name in _write (and
the other write location), enabling path traversal or invalid filenames;
sanitize both values (e.g., strip or replace path separators and restrict to a
safe character set such as alphanumerics, hyphen, underscore) before composing
the filename, and after building the path verify the resolved path is inside
self.handoff_dir (use Path.resolve() and is_relative_to or compare parents) to
refuse or re-sanitize any out-of-bound result; add a small helper like
safe_filename(...) and use it when creating the path for _write and the
duplicate write site.

---

123-127: ⚠️ Potential issue | 🟠 Major

Telemetry failure can break handoff flow and re-trigger repeatedly.

At Line 125-Line 126 and Line 143-Line 153, if events.emit(...) raises, check() fails before _fired flips. Emission should be best-effort and must not affect watchdog lifecycle.

Proposed fix

     def check(self, tokens_used: int, tokens_max: int) -> HandoffDoc | None:
@@
         doc = self.synthesizer()
         self._write(doc)
-        self._emit(pressure, doc)
         self._fired = True
+        self._emit(pressure, doc)
         return doc
@@
     def _emit(self, pressure: float, doc: HandoffDoc) -> None:
         try:
             from gradata import _events as events
-        except ImportError:
+            events.emit(
+                event_type="handoff.triggered",
+                source="handoff_watchdog",
+                data={
+                    "task_id": doc.task_id,
+                    "agent_name": doc.agent_name,
+                    "pressure": round(pressure, 3),
+                    "threshold": round(self.threshold, 3),
+                },
+                tags=["handoff", "context_pressure"],
+            )
+        except Exception:
             return
-        events.emit(
-            event_type="handoff.triggered",
-            source="handoff_watchdog",
-            data={
-                "task_id": doc.task_id,
-                "agent_name": doc.agent_name,
-                "pressure": round(pressure, 3),
-                "threshold": round(self.threshold, 3),
-            },
-            tags=["handoff", "context_pressure"],
-        )

Also applies to: 138-153

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@Gradata/src/gradata/contrib/patterns/handoff.py` around lines 123 - 127, The
handoff flow currently allows exceptions from telemetry emission to abort
check() before _fired is set; wrap calls to emit (e.g., the call site
self._emit(pressure, doc) inside check(), and any other direct events.emit
usages in the same module) in a try/except that catches and logs telemetry
errors but does not re-raise them, then ensure self._fired = True happens
regardless (use a finally or set _fired before/after in a safe block) so the
watchdog lifecycle is not affected; reference the check() method,
self.synthesizer(), self._write(), self._emit(), events.emit, and the _fired
flag when making the change.

Gradata/tests/test_handoff.py (1)

39-45: ⚠️ Potential issue | 🟠 Major

Threshold parsing tests enforce the wrong contract.

At Line 39-Line 41, the test expects fallback to 0.65 for "1.5", but contract is clamp-to-range. Update this case to assert 0.95, and add a below-min clamp case (e.g., "0.05" → 0.10).

Proposed test update

-    def test_out_of_range_falls_back(self, monkeypatch):
+    def test_out_of_range_clamps_high(self, monkeypatch):
         monkeypatch.setenv("GRADATA_HANDOFF_THRESHOLD", "1.5")
-        assert _read_threshold() == 0.65
+        assert _read_threshold() == 0.95
+
+    def test_out_of_range_clamps_low(self, monkeypatch):
+        monkeypatch.setenv("GRADATA_HANDOFF_THRESHOLD", "0.05")
+        assert _read_threshold() == 0.10

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@Gradata/tests/test_handoff.py` around lines 39 - 45, Update the threshold
parsing tests to assert clamping behavior rather than fallback: change the
expectation in test_out_of_range_falls_back (which sets
GRADATA_HANDOFF_THRESHOLD="1.5") to expect 0.95 instead of 0.65, and add a new
case (or modify an existing one) asserting that a below-min value like "0.05"
clamps to 0.10; keep the garbage input test
(GRADATA_HANDOFF_THRESHOLD="not-a-number") asserting fallback to 0.65. Ensure
references to the _read_threshold() helper remain unchanged.

🤖 Prompt for all review comments with AI agents

Verify each finding against the current code and only fix it if needed.

Duplicate comments:
In `@Gradata/src/gradata/contrib/patterns/handoff.py`:
- Around line 31-41: The _read_threshold function currently returns
_DEFAULT_THRESHOLD when an override is out-of-range or non-numeric; instead
clamp numeric overrides into the allowed range and reject only
invalid/non-finite values; update _read_threshold to parse the env var, ensure
the parsed float is finite (not NaN/Inf), then return min(max(value,
_MIN_THRESHOLD), _MAX_THRESHOLD) for valid numbers and only return
_DEFAULT_THRESHOLD for parse failures or non-finite values (use the existing
symbols _DEFAULT_THRESHOLD, _MIN_THRESHOLD, _MAX_THRESHOLD and the function
_read_threshold to locate the change).
- Around line 133-136: The filename is built directly from doc.task_id and
doc.agent_name in _write (and the other write location), enabling path traversal
or invalid filenames; sanitize both values (e.g., strip or replace path
separators and restrict to a safe character set such as alphanumerics, hyphen,
underscore) before composing the filename, and after building the path verify
the resolved path is inside self.handoff_dir (use Path.resolve() and
is_relative_to or compare parents) to refuse or re-sanitize any out-of-bound
result; add a small helper like safe_filename(...) and use it when creating the
path for _write and the duplicate write site.
- Around line 123-127: The handoff flow currently allows exceptions from
telemetry emission to abort check() before _fired is set; wrap calls to emit
(e.g., the call site self._emit(pressure, doc) inside check(), and any other
direct events.emit usages in the same module) in a try/except that catches and
logs telemetry errors but does not re-raise them, then ensure self._fired = True
happens regardless (use a finally or set _fired before/after in a safe block) so
the watchdog lifecycle is not affected; reference the check() method,
self.synthesizer(), self._write(), self._emit(), events.emit, and the _fired
flag when making the change.

In `@Gradata/src/gradata/enhancements/rag/embedders.py`:
- Around line 110-115: In embed_any(), when calling multimodal.embed(item)
(after checking multimodal.supports(item.modality)), wrap that call in a
try/except that catches NotImplementedError and allows execution to continue to
the fallback routing instead of returning immediately; specifically, catch
NotImplementedError from multimodal.embed(item), ignore or log it as
appropriate, and then fall through to the existing logic that selects
text_fallback or TextOnlyEmbedder for item.modality == "text" so runtime
rejections by a provider trigger the documented fallback behavior.

In `@Gradata/tests/test_handoff.py`:
- Around line 39-45: Update the threshold parsing tests to assert clamping
behavior rather than fallback: change the expectation in
test_out_of_range_falls_back (which sets GRADATA_HANDOFF_THRESHOLD="1.5") to
expect 0.95 instead of 0.65, and add a new case (or modify an existing one)
asserting that a below-min value like "0.05" clamps to 0.10; keep the garbage
input test (GRADATA_HANDOFF_THRESHOLD="not-a-number") asserting fallback to
0.65. Ensure references to the _read_threshold() helper remain unchanged.

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: ASSERTIVE

Plan: Pro

Run ID: 6887bc4e-e75d-431f-b999-79cad1e0e1bf

📥 Commits

Reviewing files that changed from the base of the PR and between 3698836 and 5c24a26.

📒 Files selected for processing (3)

• Gradata/src/gradata/contrib/patterns/handoff.py
• Gradata/src/gradata/enhancements/rag/embedders.py
• Gradata/tests/test_handoff.py

📜 Review details

🧰 Additional context used

🧠 Learnings (1)

📓 Common learnings

Learnt from: Gradata
Repo: Gradata/gradata PR: 0
File: :0-0
Timestamp: 2026-04-17T17:18:07.439Z
Learning: In PR `#102` (gradata/gradata), Round 2 addressed: cli.py env-first brain resolution (GRADATA_BRAIN > --brain-dir > cwd), _tenant.py corrupt .tenant_id overwrite, _env_int default clamping to minimum, and _events.py tenant-scoped fallback SELECT for dedup. All ruff and 99 tests green after these fixes.

🔇 Additional comments (1)

Gradata/tests/test_handoff.py (1)

127-160: Good regression test for event payload/signature drift.

This is a strong guardrail for handoff.triggered event shape and source semantics.

[greptile-apps]

Your free trial has ended. If you'd like to continue receiving code reviews, you can add a payment method here.

[greptile-apps]

Your free trial has ended. If you'd like to continue receiving code reviews, you can add a payment method here.

[coderabbitai]

Actionable comments posted: 4

🤖 Prompt for all review comments with AI agents

Verify each finding against the current code and only fix it if needed.

Inline comments:
In `@Gradata/src/gradata/contrib/patterns/handoff.py`:
- Around line 112-127: The check method currently sets self._fired after calling
self._emit, which can allow retries if _emit raises; change the order in
Handoff.check so that after creating doc = self.synthesizer() you set
self._fired = True before calling self._write(doc) and self._emit(pressure, doc)
(or at minimum before _emit) to ensure idempotency; ensure any exceptions from
_write/_emit are still surfaced or handled as intended but the _fired flag is
set early to prevent re-triggering.

In `@Gradata/src/gradata/hooks/inject_handoff.py`:
- Line 34: The module currently binds _MAX_HANDOFF_CHARS at import time from
GRADATA_HANDOFF_MAX_CHARS so runtime changes (e.g., in tests) are ignored;
modify the logic to read the environment variable lazily instead: remove or stop
using the module-level _MAX_HANDOFF_CHARS constant and update _read_threshold
(or whichever call sites use _MAX_HANDOFF_CHARS) to call
os.environ.get("GRADATA_HANDOFF_MAX_CHARS", "4000") and parse int at call time
so changes to GRADATA_HANDOFF_MAX_CHARS after import are respected.
- Around line 63-67: The XML attribute for source is built directly from
candidate.name in the handoff block (see the f'<handoff
source="{candidate.name}">' construction) and must be escaped to prevent
injection; replace the raw candidate.name with an escaped version (use a
standard XML/HTML escaping utility such as html.escape or
xml.sax.saxutils.escape that also escapes quotes) before interpolating, so the
resulting block is safe even if the filename contains characters like " or <.

In `@Gradata/tests/test_inject_handoff_hook.py`:
- Around line 137-140: Replace the list comprehension + indexing used to extract
the single "handoff.injected" call with a next(...) expression so the test fails
fast: use next(c for c in calls if c[0] == "handoff.injected") to get the
matching call tuple and then access its payload (the second element) to assign
to injected; keep the preceding existence assertion or remove it since next will
raise StopIteration if not found, and ensure the rest of the assertions still
check injected["file"] and injected["chars"].

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: ASSERTIVE

Plan: Pro

Run ID: c2e0ca13-f3e4-4f3b-9a31-ae09bec555e2

📥 Commits

Reviewing files that changed from the base of the PR and between b49b66f and 43a1905.

📒 Files selected for processing (4)

• Gradata/src/gradata/contrib/patterns/handoff.py
• Gradata/src/gradata/hooks/inject_handoff.py
• Gradata/tests/test_handoff.py
• Gradata/tests/test_inject_handoff_hook.py

📜 Review details

🧰 Additional context used

🧠 Learnings (1)

📓 Common learnings

Learnt from: Gradata
Repo: Gradata/gradata PR: 0
File: :0-0
Timestamp: 2026-04-17T17:18:07.439Z
Learning: In PR `#102` (gradata/gradata), Round 2 addressed: cli.py env-first brain resolution (GRADATA_BRAIN > --brain-dir > cwd), _tenant.py corrupt .tenant_id overwrite, _env_int default clamping to minimum, and _events.py tenant-scoped fallback SELECT for dedup. All ruff and 99 tests green after these fixes.

🪛 Ruff (0.15.10)

Gradata/tests/test_inject_handoff_hook.py

[warning] 138-138: Prefer next(c for c in calls if c[0] == "handoff.injected") over single element slice

Replace with next(c for c in calls if c[0] == "handoff.injected")

(RUF015)

🔇 Additional comments (15)

Gradata/src/gradata/contrib/patterns/handoff.py (5)

31-41: Threshold contract: still falls back instead of clamping.

The PR description states the threshold is "clamped to [0.10, 0.95]", but this implementation returns the default 0.65 for out-of-range values instead of clamping to the nearest bound. Additionally, nan would slip through since nan < 0.10 and nan > 0.95 are both False.

Suggested fix per documented contract

+import math
+
 def _read_threshold() -> float:
     raw = os.environ.get("GRADATA_HANDOFF_THRESHOLD", "")
     if not raw:
         return _DEFAULT_THRESHOLD
     try:
         value = float(raw)
     except ValueError:
         return _DEFAULT_THRESHOLD
-    if value < _MIN_THRESHOLD or value > _MAX_THRESHOLD:
+    if not math.isfinite(value):
         return _DEFAULT_THRESHOLD
-    return value
+    return max(_MIN_THRESHOLD, min(_MAX_THRESHOLD, value))

---

133-136: Path traversal vulnerability: unsanitized task_id and agent_name in filename.

These values are used directly in the filename. Inputs like task_id="../../../etc" or agent_name="foo/bar" could write files outside handoff_dir or create invalid paths.

Suggested fix

+import re
+
+def _safe_name(value: str) -> str:
+    """Sanitize a value for use as a filename component."""
+    cleaned = re.sub(r"[^A-Za-z0-9._-]+", "_", value).strip("._")
+    return cleaned[:64] or "handoff"  # length limit too
+
     def _write(self, doc: HandoffDoc) -> None:
         self.handoff_dir.mkdir(parents=True, exist_ok=True)
-        path = self.handoff_dir / f"{doc.task_id}_{doc.agent_name}.handoff.md"
+        path = self.handoff_dir / f"{_safe_name(doc.task_id)}_{_safe_name(doc.agent_name)}.handoff.md"
         path.write_text(doc.render(), encoding="utf-8")

The same sanitization should be applied in load_handoff (lines 156-158).

---

138-153: Telemetry emission can fail and break the watchdog.

The events.emit() call at line 143 is outside the try/except block. If emit() raises an exception, check() will throw after the handoff file is written but before _fired is set to True. This causes:

1. User-facing failure from an observability call
2. Repeated re-triggers on subsequent check() calls

Suggested fix

     def _emit(self, pressure: float, doc: HandoffDoc) -> None:
         try:
             from gradata import _events as events
+            events.emit(
+                event_type="handoff.triggered",
+                source="handoff_watchdog",
+                data={
+                    "task_id": doc.task_id,
+                    "agent_name": doc.agent_name,
+                    "pressure": round(pressure, 3),
+                    "threshold": round(self.threshold, 3),
+                },
+                tags=["handoff", "context_pressure"],
+            )
-        except ImportError:
+        except Exception:
             return
-        events.emit(
-            event_type="handoff.triggered",
-            source="handoff_watchdog",
-            data={
-                "task_id": doc.task_id,
-                "agent_name": doc.agent_name,
-                "pressure": round(pressure, 3),
-                "threshold": round(self.threshold, 3),
-            },
-            tags=["handoff", "context_pressure"],
-        )

---

193-208: LGTM on pick_latest_unconsumed implementation.

The function correctly ignores the consumed/ subdirectory by checking p.parent == handoff_dir, sorts by mtime descending, and handles edge cases (missing dir, empty dir).

---

177-190: LGTM on consume_handoff design.

Silent failure is appropriate here since injection already succeeded. Preserving files in consumed/ for audit is a good practice.

Gradata/tests/test_handoff.py (5)

42-48: Tests encode fallback behavior, but PR describes clamping.

The PR objectives state the threshold is "clamped to [0.10, 0.95]", but these tests assert fallback to 0.65 for out-of-range values. Either the implementation should be updated to clamp (and tests updated accordingly), or the documentation should clarify the actual behavior.

Additionally, consider adding tests for:

• Below-min threshold (e.g., 0.01 → expected 0.10 if clamping, 0.65 if fallback)
• NaN input (currently slips through the range check)

---

80-95: LGTM on helper factory method.

The _make helper cleanly encapsulates watchdog setup for reuse across tests.

---

130-163: LGTM on emission test.

Good coverage of the handoff.triggered event payload including task_id, agent_name, threshold, pressure range, and tags.

---

202-212: LGTM on mtime-based selection test.

Using os.utime to set the older file's timestamp in the past is a reliable way to test mtime-based sorting.

---

235-236: LGTM on silent failure test for missing file.

This confirms consume_handoff doesn't raise when the source file doesn't exist.

Gradata/src/gradata/hooks/inject_handoff.py (2)

1-11: LGTM on module docstring and design.

The docstring clearly explains the hook's purpose, ordering (runs before brain-rules for primacy), consumption semantics, and skip policy. This aligns well with the PR objectives.

---

71-81: LGTM on best-effort event emission.

The try/except wrapper ensures telemetry failures don't break the hook. Debug-level logging is appropriate.

Gradata/tests/test_inject_handoff_hook.py (3)

10-15: LGTM on fixture design.

The fixture correctly sets up the brain directory environment and creates the handoffs subdirectory. Returning both paths is convenient for tests.

---

75-84: Test requires module reload due to module-level constant.

The importlib.reload is necessary because _MAX_HANDOFF_CHARS is evaluated at import time. This is a valid workaround, though it couples the test to the implementation detail.

---

143-149: LGTM on missing brain directory test.

Good coverage ensuring the hook gracefully returns None when no brain directory is resolvable.

[coderabbitai]

🧹 Nitpick | 🔵 Trivial

Consider setting _fired = True before _emit to ensure idempotency.

If _emit fails (even with the fix above), the current flow sets _fired after _emit. Moving it earlier ensures the watchdog won't re-trigger even if emission has issues.

Proposed reordering

         doc = self.synthesizer()
         self._write(doc)
+        self._fired = True
         self._emit(pressure, doc)
-        self._fired = True
         return doc

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@Gradata/src/gradata/contrib/patterns/handoff.py` around lines 112 - 127, The
check method currently sets self._fired after calling self._emit, which can
allow retries if _emit raises; change the order in Handoff.check so that after
creating doc = self.synthesizer() you set self._fired = True before calling
self._write(doc) and self._emit(pressure, doc) (or at minimum before _emit) to
ensure idempotency; ensure any exceptions from _write/_emit are still surfaced
or handled as intended but the _fired flag is set early to prevent
re-triggering.

[coderabbitai]

🧹 Nitpick | 🔵 Trivial

Environment variable evaluated at import time.

_MAX_HANDOFF_CHARS is read once when the module is imported. If a test or caller changes GRADATA_HANDOFF_MAX_CHARS after import, the change won't take effect without reloading the module. This is consistent with how _read_threshold() works in the handoff module, but worth noting for test authors (as seen in test_truncates_oversized_body which does importlib.reload).

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@Gradata/src/gradata/hooks/inject_handoff.py` at line 34, The module currently
binds _MAX_HANDOFF_CHARS at import time from GRADATA_HANDOFF_MAX_CHARS so
runtime changes (e.g., in tests) are ignored; modify the logic to read the
environment variable lazily instead: remove or stop using the module-level
_MAX_HANDOFF_CHARS constant and update _read_threshold (or whichever call sites
use _MAX_HANDOFF_CHARS) to call os.environ.get("GRADATA_HANDOFF_MAX_CHARS",
"4000") and parse int at call time so changes to GRADATA_HANDOFF_MAX_CHARS after
import are respected.

[coderabbitai]

⚠️ Potential issue | 🟡 Minor

Escape candidate.name in the XML attribute to prevent injection.

If a handoff filename contains characters like " or <, the generated XML block could break or be exploited. Consider escaping the source attribute value.

Proposed fix

+import html
+
 ...
 
     safe = _sanitize(body.strip())
-    block = f'<handoff source="{candidate.name}">\n{safe}\n</handoff>'
+    block = f'<handoff source="{html.escape(candidate.name, quote=True)}">\n{safe}\n</handoff>'

📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change

	if len(body) > _MAX_HANDOFF_CHARS:
	body = body[:_MAX_HANDOFF_CHARS] + "\n<!-- truncated -->"
	safe = _sanitize(body.strip())
	block = f'<handoff source="{candidate.name}">\n{safe}\n</handoff>'
	if len(body) > _MAX_HANDOFF_CHARS:
	body = body[:_MAX_HANDOFF_CHARS] + "\n<!-- truncated -->"
	safe = _sanitize(body.strip())
	block = f'<handoff source="{html.escape(candidate.name, quote=True)}">\n{safe}\n</handoff>'

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@Gradata/src/gradata/hooks/inject_handoff.py` around lines 63 - 67, The XML
attribute for source is built directly from candidate.name in the handoff block
(see the f'<handoff source="{candidate.name}">' construction) and must be
escaped to prevent injection; replace the raw candidate.name with an escaped
version (use a standard XML/HTML escaping utility such as html.escape or
xml.sax.saxutils.escape that also escapes quotes) before interpolating, so the
resulting block is safe even if the filename contains characters like " or <.

[coderabbitai]

🧹 Nitpick | 🔵 Trivial

Use next() instead of list slicing for single element extraction.

Per static analysis (RUF015), prefer next(c for c in calls if c[0] == "handoff.injected") for clarity and to fail fast if no match exists.

Proposed fix

         inject_handoff.main({})
         assert any(c[0] == "handoff.injected" for c in calls)
-        injected = [c for c in calls if c[0] == "handoff.injected"][0][1]
+        injected = next(c for c in calls if c[0] == "handoff.injected")[1]
         assert injected["file"] == "x.handoff.md"
         assert injected["chars"] > 0

📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change

	assert any(c[0] == "handoff.injected" for c in calls)
	injected = [c for c in calls if c[0] == "handoff.injected"][0][1]
	assert injected["file"] == "x.handoff.md"
	assert injected["chars"] > 0
	assert any(c[0] == "handoff.injected" for c in calls)
	injected = next(c for c in calls if c[0] == "handoff.injected")[1]
	assert injected["file"] == "x.handoff.md"
	assert injected["chars"] > 0

🧰 Tools

🪛 Ruff (0.15.10)

[warning] 138-138: Prefer next(c for c in calls if c[0] == "handoff.injected") over single element slice

Replace with next(c for c in calls if c[0] == "handoff.injected")

(RUF015)

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@Gradata/tests/test_inject_handoff_hook.py` around lines 137 - 140, Replace
the list comprehension + indexing used to extract the single "handoff.injected"
call with a next(...) expression so the test fails fast: use next(c for c in
calls if c[0] == "handoff.injected") to get the matching call tuple and then
access its payload (the second element) to assign to injected; keep the
preceding existence assertion or remove it since next will raise StopIteration
if not found, and ensure the rest of the assertions still check injected["file"]
and injected["chars"].

[Gradata]

Superseded by main — all novel work (handoff watchdog, multimodal RAG embedder, meta_rules port, context_inject, rag_embedders) has already landed via #102, #121, #133, and feat/token-optimization-autoresearch. Verified: after merging main into this branch with main winning conflicts, diff vs main is empty. Closing.