thin-channel-thick-runner: strangler-fig collection refactor + AuthManager + per-domain limits - #3
Merged
Merged
Conversation
…llection network Node-kit (frontend/src/node-kit): spec/define/registry contract, generic xyflow KitNode renderer, L3 atom components, 19 real atomic nodes (sources/processors/ pipeline/primitives), agent JSON-schema bridge. ComfyUI-style NodeWorkbench (left palette + dnd-kit drag, Tab/double-click search, collision animation) at /labs/node-kit. ELK auto-layout: render/elkLayout.ts (elkjs layered, direction RIGHT) + ambient elk.d.ts (elkjs ships no types); "auto-layout" toolbar button snaps a scattered graph into a clean dataflow then fitView. Integration: NetworkPage atomMode toggle maps a collection source into an atom seed graph (L1 project -> L2 stage -> L3 atoms). Self-built P0 runtime (runtime/engine.ts: Kahn topo-sort, pure-node run, backend hook). Also: agent dock (backend/api/v1/chat.py + frontend AgentDock), nav/i18n/Layout wiring, backend pipeline runner tweaks.
Step 1 of unifying 采集网络 and the node-kit workbench into one engine, two
views. The topology canvas stage cards now render through the generic <KitNode>
instead of a hand-rolled TopologyNodeView — both surfaces share one node
language (same registry, same specs).
- node-kit/nodes/collection.tsx (new): 8 collection.* specs (source/schedule/
task/agent/record/notification/edge-node/worker). Specs with actions carry a
shared StageBody render (status/gap facts + badge chips, matching the old
card) and ops calling the same endpoints StageOperations uses.
- ReactFlowTopologyCanvas: dropped TopologyNodeView + static nodeTypes + dead
health-class helpers; registers ALL_NODES, memoizes nodeTypesForXyflow(), and
the sync effect re-types nodes to collection.<kind> with a data superset
({...data, config, facts}) so the right-drawer inspector keeps working.
- Guard unknown kinds (hasNode fallback) so a new backend kind can't silently
vanish; badges coerced with String() instead of an unsound cast.
NetworkPage atomMode + StageOperations untouched. tsc green; verified in
browser: 采集网络 renders the source stage via collection.source, node-kit
palette now lists all 8 collection nodes (unified language). Review false-
positive noted: stageConfig enabled derives from health, which topologyModel
builds from enabled (health==='disabled' iff !enabled) — toggle is correct.
…node Step 2 of unifying 采集网络 (macro/zoom-out view) and the workbench (atom/zoom-in view). A macro = a saved atom subgraph that collapses to one node and expands back. Iron rule honored: NO second executor — runGraph stays the only engine; a macro is flattened to its atoms before a run. NEW node-kit/macros/: - macro.ts: MacroDef/MacroPort, deriveBoundaryPorts (a handle is a boundary iff it has no internal edge, keyed like the engine: targetHandle??'in' / sourceHandle??'out'), buildMacroDef, makeMacroSpec (synthetic NodeSpec per macro via defineNode, ports id = innerNodeId:innerHandle), inlineMacro (pure namespace-by-instance-id + offset + reconnect), flattenForRun (loop-inline, guard 50). In-memory MACRO_DEFS map + getMacroDef for the hot paths. - store.ts: localStorage 'node-kit:macros' (list/save/get/delete) + isMacroDef guard + try/catch; strips transient node fields before persist. - MacroNode.tsx: collapsed body (child-type chips + 双击展开 hint). - index.ts: barrel + registerSavedMacros() (boot-time re-register). NodeWorkbench: 组成宏 button (>=2 selected) captures selection -> derives ports -> saves/registers -> replaces selection with one macro node, rewiring crossing edges onto synthetic ports; double-click a macro expands it back; runNow flattens first; nodeTypes/palette memos keyed on a registryVersion so an in-session macro renders + lists immediately. Nested macros refused (MVP bound). NodeKitPage/NetworkPage/ReactFlowTopologyCanvas register saved macros at boot. Review fixes applied: single atomic inlineMacro on expand (was a torn two-call update); createElement(MacroBody) instead of a plain call (hooks-safe); getMacroDef in-memory lookup replaces localStorage JSON.parse on the run/expand hot path; saveMacro logs on quota failure. tsc green; /labs/node-kit loads clean (组成宏 button present, no console errors).
Node config was display-only — "里面东西不能改". Make the spec-driven node body editable in place, ComfyUI-style. - atoms: NodeFieldEdit renders one FieldDef as the right control by type — text / number / select / json(textarea) / boolean(NodeToggle). Carries `nodrag nopan` + a pointerdown guard so editing never drags the node or pans the canvas; half-typed JSON is kept as raw text rather than lost. - KitNode: AutoBody now renders config.fields as editable NodeFieldEdit controls (facts stay read-only) and writes changes back via useReactFlow().updateNodeData onto this node's data.config, so edits flow into runGraph. Only affects nodes WITHOUT a custom spec.render — collection.* (StageBody) and macro nodes keep their own bodies, so 采集网络 stays backend-authoritative. tsc green; verified in browser: web_scraper/processor nodes render text/number/ select/toggle inputs, typing into a field persists through re-render (controlled value bound to data.config — proves write-back), zero console errors.
Two asks: node config editing should be more convenient (humans), and atomic
nodes more convenient for AI development.
1. Side property panel (render/NodeInspector.tsx): selecting ONE node shows a
roomy right-side form of its full config (same NodeFieldEdit controls as the
inline body, bigger + always visible). Writes go through updateNodeData, so
panel and inline edits stay in sync. Works for any node with config.fields
incl. collection.* whose compact body is StageBody.
2. AI graph authoring API (agent/graph.ts): instantiateGraph({nodes,edges}) —
the inverse of agent/toSchema. Validates an agent-emitted graph against the
registry (unknown types, config errors, dangling edges, bad port refs all
collected, never thrown), returns canvas-ready xyflow nodes+edges + a list of
problems so the agent can fix-and-retry. A fuchsia "AI 产图" toolbar button
loads pasted graph JSON onto the canvas with ELK fit + error toast.
tsc green; verified in browser: AI 产图 with a 4-node/3-edge blob (one unknown
type, one dangling edge) loads exactly the 3 valid nodes + 2 valid edges,
rejects the rest; clicking a node opens the property panel with its editable
config; zero console errors.
Closed-loop browser-skill subsystem (ADR-0003): a `skill` channel reads a distilled SKILL.md and a cheap text model drives a real Chrome page over CDP through a perceive -> gate -> act loop, staying inside the existing task/run/pipeline/events/record spine. - backend/skills/: distill kernel, Playwright connect_over_cdp page wrapper, injected-JS perception snapshot, ref-addressed action executor, step loop with 9-element prompt + tool-calling harness, risk-tiered confirm gate, journey_trace_v1 emission + self-eval + re-distill correction - models/Skill with (domain, capability) unique; awaiting_confirm run status - skill_channel wired into the pipeline (run_id via parameters, per-step events, extract -> records); AbstractChannel.collect signature unchanged - migrations m3h4i5j6k7l8 (skills), n4i5j6k7l8m9 (awaiting_confirm); /skills API - tests/skills (browser-free unit) + live e2e behind a `live` marker - fix 2 pre-existing async-mock failures in tests/unit/test_runner.py - docs: ADR-0003, GLOSSARY, PRD, per-issue specs Built via grilled design (/grill-with-docs) + multi-agent implementation workflow.
Two normalized triples can share a content_hash (e.g. two CLI sub-commands that normalize to identical content). Both passed the existing-hashes check and were added, failing the whole batch on the UNIQUE(source_id, content_hash) constraint at flush. Track hashes seen in this batch too.
Wire the dock correct leg (ADR-0003 D7/D8): when the context node is a
failing skill, surface a 重蒸技能 button that opens an amber confirm card
and, on confirm, POSTs the failing journey_trace_v1 to
/skills/{id}/redistill. Reuses the existing proposal->confirm contract and
never auto-fires (D8: re-distill is human-triggered only). Pairs with the
backend endpoint (api/v1/skills.py) + correction.re_distill from 3bb827a.
- RedistillTarget state + failingTrace prop (falls back to a minimal
context-only trace so the flow is exercisable without a run)
- propose/confirm/cancel handlers with loading guards + toast/append feedback
Close the execute-from-store seam (ADR-0003). The skill channel could only run an inline config['skill_md']; a skill_id / (domain, capability) was rejected with "resolution not wired yet", so a distilled skill in the DB could not be executed (blocking end-to-end QA). _resolve_skill now loads the persisted Skill via a short-lived AsyncSessionLocal — the same pattern as the self-eval evidence write, since collect() holds no injected session. - _load_skill_fields: read-only load by skill_id then the unique (domain, capability); reads columns inside the session and returns a plain dict, so the caller never touches a detached ORM instance - _resolve_skill: inline skill_md still wins (fast path); else resolve from DB, guarding disabled / empty-body / not-found with clean ChannelResult.fail - resolved identity (skill_id/domain/capability/version) flows into the journey_trace + self-eval write-back - tests: resolve by skill_id and by (domain, capability), disabled refusal, unknown-skill clean failure (whole skills suite: 96 passed)
Make backend/skills/ a self-contained, reusable package — importing the execute loop no longer drags in the FastAPI dock or the pipeline/DB spine. - new backend/skills/toolcall.py: the pure tool-call parse helpers (_is_xml_tool_model / _parse_tool_use / _safe_json + their constants), owned by the skills package instead of api.v1.chat. Breaks the skills.loop -> api.v1.chat import cycle (the lazy-import workarounds in skill_channel existed only because of it). - loop.py imports those helpers from skills.toolcall and takes an injected `emit` sink (default None / no-op) instead of importing backend.pipeline.events, removing the loop's last spine dependency. skill_channel passes emit=events.emit, so run-event behaviour is unchanged. - proof: `import backend.skills.loop` pulls in neither backend.api.v1.chat nor backend.pipeline.events; tests/skills green (96 passed, 2 live deselected). Residual (host-side, does not block reuse): api.v1.chat still defines its own copy of the three helpers — collapse to `from backend.skills.toolcall import ...` when that (currently WIP) file is next committed. correction.py / distill.py still touch ORM models at the adapter edge.
… entry into the skill execute domain
backend/api/v1/skill_bridge.py: a thin, domain-neutral mapper around SkillChannel.collect
for the universal-studio kernel's PythonBridge transport. Honors the cross-language wire
envelope { capability, params, inputs } -> { ok, outputs:{records,trace,self_eval}, events,
error? }; outputs are typed (records DataRef<Record>, trace DataRef<JourneyTrace>, self_eval
Value<SelfEval>) mapped from ChannelResult items+metadata; events project the journey trace
steps (post-hoc node.progress). Own router (NOT the agent dock chat.py), registered in
api/v1/__init__.py. Test via existing browser/model patch fixtures: tests/skills 97 passed.
Verified live end-to-end: real qwen3:4b (Ollama) drove real Chrome (CDP :9222) through the
real SkillChannel, returning extracted records + journey trace back through the kernel's
PythonBridge with node.progress events.
Pairs with universal-studio commit ed38c7e (PythonBridge TS half).
…FetchResult/Capabilities) North star: adding a real data source should be ~100 lines of source-specific "send one request, parse the response into items". Every cross-cutting concern — token refresh, pagination, rate limiting, cursor persistence — belongs to the runner, not the channel. Today they are inlined per-channel (opencli's collect() is ~469 lines), so each source either reimplements them or can't do them. This work flips it: thin channels that declare capabilities + source logic, a thick runner that owns the cross-cutting concerns once. Phase 0 is purely additive and non-breaking: it introduces the thick contract and lets the existing channels inherit it via a default adapter. No runtime path changes, no behaviour change. - backend/channels/base.py: new Capabilities (frozen — incremental / paginated / auth_kind / session_affinity / default_rate), AuthContext (Phase 2 placeholder), FetchContext (context in), FetchResult (items + next_cursor + has_more), and ChannelFetchError. AbstractChannel gains a default `capabilities`, a default `fetch(ctx)` that bridges to the legacy collect() (one-shot, no cursor), and a default `identity(item) -> str | None` (None → the normalizer keeps its content hash, so dedup is unchanged this phase). - collect() stays the abstract method, so the six existing channels are unchanged and inherit the contract for free. The adapter lives once in the base class (locality), not as six wrappers. - pipeline / normalizer untouched — the new hooks are ignored by the runtime path this phase. identity() wiring, the runner three-piece (cursor store + retry/ backoff client + token bucket), and RSS etag land in Phase 1. tests/unit/channels/test_contract.py proves the seam: a collect-only channel (the shape of all six) gets fetch()/identity()/capabilities for free, a failed collect surfaces as ChannelFetchError, and RSSChannel inherits the contract unchanged. All channel unit tests green (117 passed).
…te-limited retrying client + pagination)
The cross-cutting concerns a channel should never reimplement, built once for the
runner to own. Additive and not yet wired into the live collect stage
(collector.py still calls channel.collect); the DB-backed cursor store + migration,
the RSS etag override, and the pipeline switch are the next slice.
- backend/pipeline/http_client.py: TokenBucket (async, burst-aware) + parse_rate
("60/min" -> tokens/s) + RateLimitedClient wrapping httpx.AsyncClient — every
request waits on the bucket then retries 429/5xx with exponential backoff +
jitter, honoring a numeric Retry-After. A channel does `await ctx.http.get()` and
gets all of it for free.
- backend/pipeline/cursor_store.py: CursorStore Protocol + InMemoryCursorStore. The
runner depends on the Protocol (accept dependencies, don't create them); the
DB-backed adapter swaps in behind it with no change above.
- backend/pipeline/channel_runner.py: run_channel(source, params) — loads the
cursor (when the channel is incremental), builds the rate-limited client from the
channel's declared rate, drives fetch() through pagination via
has_more/next_cursor, persists the cursor after each page (crash mid-pagination
resumes, not restarts), and guards with MAX_PAGES. channel/http/cursor_store are
injectable for tests.
tests/unit/pipeline: run_channel drives pagination + saves a cursor per page +
resumes from a stored cursor + runs a collect-only channel once + honors MAX_PAGES;
the client retries 429->200, honors Retry-After, and gives up after max_retries.
All unit tests green (168 passed across pipeline + channels).
…PR1) Insert an ItemSink seam between collection and the write destination, so a source's data can later flow to the ODP hot path (OdpSink) or both at once (DualSink) by selecting a sink — never by rewriting the pipeline. First cut of the strangler-fig migration: make the boundary replaceable without changing what crosses it. This PR is behavior-preserving. LegacyDbSink wraps the existing normalizer + storer path, INCLUDING the storer-level ODP forward that already fires when ODP_INGEST_URL is configured. Extracting that forward into OdpSink (and gating LegacyDbSink so DualSink cannot double-send) is intentionally deferred to PR3 — LegacyDbSink is therefore not yet a pure legacy-DB sink. - backend/pipeline/sinks/base.py: ItemSink Protocol + RunContext + SinkResult. SinkResult.records carries the persisted ORM rows so the downstream AI/notify steps keep working unchanged; its count semantics (accepted/duplicates/ rejected) are pinned per-sink relative to each sink's own durable boundary. - backend/pipeline/sinks/legacy_db_sink.py: the original normalize+store path, moved behind the seam with no behavior change. Carries a PR3 TODO at the storer call for the forward_to_odp gate. - backend/pipeline/pipeline.py: steps 2+3 now delegate to active_sink.write_batch; run_pipeline gains an injectable `sink=` (defaults to LegacyDbSink). Tests: LegacyDbSink normalizes then stores and maps the result; run_pipeline delegates through an injected sink end-to-end. Existing test_pipeline / test_storer / test_normalizer stay green = the behavior-unchanged proof. All tests/unit green.
…rangler-fig PR2) Pin the wire shape opencli-admin forwards to the Rust ingest service so a later step can move the forward out of storer into an OdpSink behind a characterization test proving equivalence. Forward is NOT moved yet — that is PR3. - backend/odp/schemas.py: RecordEvent / OdpIngestResponse / IngestReject, a typed mirror of odp-rs/crates/odp-contracts (SCHEMA_VERSION=1). to_wire() reproduces the legacy forwarder bytes exactly (explicit nulls, stringified ids); now also parses the response `errors` array the old code dropped. - backend/odp/mapper.py: RecordEventMapper — normalized record -> RecordEvent, single source of truth for the ODP payload shape (input is the normalized record, not the raw collector item). - backend/pipeline/odp_client.py: triple_to_event delegates to the mapper; post_batch parses OdpIngestResponse. Public signatures unchanged; storer untouched. - tests: pin the literal wire dict, mapper field mapping, and the storer forward gate (url set/unset, fail-open default, ODP_INGEST_REQUIRED fail-closed). 347 passed (was 325).
…(strangler-fig PR3) The double-send trap: storer.store_records forwards to ODP on its own when ODP_INGEST_URL is set. A naive DualSink(LegacyDbSink + OdpSink) would then send each batch to ODP twice, polluting the shadow comparison. This slice resolves it without breaking the legacy path. - storer.store_records: add forward_to_odp gate (default True = behavior unchanged); the env-driven forward only fires when the flag is on. - LegacyDbSink(forward_to_odp=True): threads the gate to storer. - OdpSink: forward-only sink — normalizes, posts via the PR2 mapper/client, SinkResult.records=[] so AI/notify no-op on the ODP leg. Propagates failures (odp_primary/odp_only need to see them). - DualSink: legacy write (authoritative, forward_to_odp=False) + OdpSink shadow forward exactly once; ODP failure is logged + recorded in SinkResult.errors and never blocks the legacy write. write_strategy selection of these sinks is PR4 — default pipeline still uses LegacyDbSink(), behavior unchanged. 355 passed (was 347).
…angler-fig PR4)
Pipeline write destination is now chosen per-source by a declared strategy
instead of an implicit env-var side effect. An injected sink still wins (tests,
callers); otherwise select_sink(source.write_strategy) decides.
- data_sources.write_strategy column (default 'legacy') + alembic migration
o5j6k7l8m9n0 (server_default='legacy' so existing rows keep current behavior).
- backend/pipeline/sinks/strategy.py select_sink:
legacy -> LegacyDbSink() (DB + original env-gated shadow)
odp_shadow -> DualSink(require_odp=False)(DB authoritative, ODP best-effort once)
odp_dual_required -> DualSink(require_odp=True) (ODP failure surfaced)
odp_primary -> DualSink(require_odp=True) (write path == dual_required;
read-routing out of scope)
odp_only -> OdpSink() (no DB row)
unknown/None -> legacy (warns)
- DualSink gains require_odp: re-raise on ODP failure instead of swallow.
- pipeline.py wires select_sink at the write seam.
Default 'legacy' = behavior unchanged. 365 passed (was 355).
…r-fig PR5a)
The additive building blocks for the RSS vertical slice — all behind the
existing thick-channel seam, so the live pipeline (still on collect()) is
untouched. The collect-stage cutover to run_channel() is PR5b.
- source_cursors table (model + alembic p6k7l8m9n0o1, one row per source).
- DBCursorStore: CursorStore Protocol backed by source_cursors, upsert on save,
own short-lived session (mirrors the sinks).
- RSSChannel migrated onto the thick contract:
* capabilities = incremental (resumes from a persisted cursor)
* fetch(): conditional GET via the cursor's etag/last_modified — 304 keeps the
cursor and returns no items; 200 reparses and advances the cursor to the
response ETag/Last-Modified. Uses ctx.http (rate-limited) when present.
* identity() = entry id — a stable dedup key (edited title != new item).
- test_contract: RSS is now a migrated channel, so the "inherits defaults"
witness moves to CLIChannel (still collect-only).
374 passed (was 365).
…committed post-write (strangler-fig PR5b) The live cutover, opt-in by capability: only channels declaring capabilities.incremental (RSS today) route through the thick runner; every other channel keeps the unchanged one-shot collect() path. - collector.collect: incremental channels go through run_channel with an in-memory staging cursor seeded from the DB cursor. The advanced cursor rides back in metadata['cursor_pending'] instead of being persisted during fetch. - pipeline: after the write sink accepts the batch (a failed sink returned earlier), commit the staged cursor to DBCursorStore. The cursor never advances past data that did not durably land; committing during fetch would skip unwritten items. Deeper ODP durability (a queued 202 that never persists) is an ODP-side guarantee, tracked separately. - test_collector: the dispatch mocks now declare Capabilities() (non-incremental) so they exercise the legacy path explicitly. Non-incremental behavior unchanged. 379 passed (was 374).
…nity (Phase 3 PR-A)
Generalize the chrome-binding pre-step: instead of a hardcoded
`channel_type in ("opencli", "skill")`, the pipeline now gates on the channel's
declared `capabilities.session_affinity`. A new session-bound channel needs no
edit to the pipeline.
- OpenCLIChannel / SkillChannel declare Capabilities(session_affinity=True).
- pipeline.py resolves the channel via the registry and checks the capability
(unknown channel_type still surfaces in the collect step, unchanged).
Behavior-preserving: the same two channels are gated as before. 382 passed (was 379).
Secrets stop living as inline plaintext in channel_config: store them encrypted
and resolve them at runtime into the runner's AuthContext, so channels never
touch raw values.
- backend/auth/crypto.py: Fernet wrapper, master key from env
CREDENTIAL_ENCRYPTION_KEY (read lazily; encrypt/decrypt raise a clear
CredentialCryptoError on missing/invalid key or corrupt token).
- source_credentials table (model + alembic q7l8m9n0o1p2): ciphertext only, one
row per (source_id, key_name).
- backend/auth/manager.py AuthManager: store() encrypts+upserts; resolve()
decrypts to {key_name: value}; resolve_context(source_id, auth_kind) shapes
bearer/api_key/basic into AuthContext (auth_kind="none" short-circuits, no DB).
- channel_runner: fills AuthContext via AuthManager.resolve_context (replaces the
Phase-0 placeholder); RSS (auth_kind=none) path unchanged, no DB hit.
- api_channel: logs a deprecation warning when an inline plaintext token/key/
password is used; resolved header is byte-identical. Env indirection
(token_env / {{secret:ENV}}) stays quiet.
- cryptography promoted to a direct dependency.
Wiring depth stops at AuthManager (api_channel is not forced onto fetch() — that
is a follow-up). 397 passed (was 382).
…R-B) Bound how many collection runs touch the same host at once, so the fleet stays polite to a site even when many sources target it. Enforced around the pipeline run in run_collection_pipeline, so it covers every channel type — including the browser-driven opencli/skill channels that never go through run_channel. - backend/pipeline/domain_limiter.py: domain_of(source) derives the host from the channel_config (feed_url/base_url/url/site/endpoint); domain_slot() is an async per-domain semaphore (limit from PER_DOMAIN_CONCURRENCY, default 3). No-op when no domain can be derived (e.g. cli). The registry is keyed by (loop, domain) so a semaphore is never reused across event loops. - runner.py wraps Phase 3 in `async with domain_slot(source)`. In-process cap (one worker); strict cross-worker limiting would swap a Redis limiter behind the same call site. 404 passed (was 397).
|
Important Review skippedToo many files! This PR contains 238 files, which is 88 over the limit of 150. To get a review, narrow the scope: Upgrade to a paid plan to raise the limit. ⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: ⛔ Files ignored due to path filters (3)
📒 Files selected for processing (238)
You can disable this status message by setting the Use the checkbox below for a quick retry:
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces the 'Skill' execute loop subsystem, enabling browser automation over CDP using Playwright, along with encrypted credential storage, incremental cursors, and a decoupled write sink architecture supporting both legacy database writes and ODP hot-path forwarding. On the frontend, Vite is established as the sole production mainline. The review feedback identifies several key issues: a regex bug in parsing nested JSON payloads within tool calls, database query inefficiencies in event polling and skill counting, tight coupling in the dual-write sink on ODP failures, potential flakiness on dynamic SPAs due to virtual DOM re-renders, a lack of double-quote escaping in prompt generation, and missing type validation on traces during re-distillation.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
Comment on lines
+367
to
+369
| _TOOL_USE_RE = re.compile( | ||
| r'<tool_use\s+name="([^"]+)"[^>]*?(?:/\s*>|>\s*(\{.*?\}|)\s*</tool_use>)', re.DOTALL | ||
| ) |
There was a problem hiding this comment.
The non-greedy brace matching \{.*?\} in the _TOOL_USE_RE regular expression will stop at the first closing brace }. If the JSON payload contains nested objects (e.g., {"options": {"cron": "..."}}), this regex will truncate the JSON string, causing json.loads to fail with a JSONDecodeError (which then returns an empty dict {} via _safe_json).\n\nSince the JSON is enclosed by the </tool_use> tag, we can safely match any characters up to the closing tag using (.*?) instead.
_TOOL_USE_RE = re.compile(\n r'<tool_use\\s+name="([^"]+)"[^>]*?(?:/\\s*>|>\\s*(.*?)\\s*</tool_use>)', re.DOTALL\n)
Comment on lines
+28
to
+30
| _TOOL_USE_RE = re.compile( | ||
| r'<tool_use\s+name="([^"]+)"[^>]*?(?:/\s*>|>\s*(\{.*?\}|)\s*</tool_use>)', re.DOTALL | ||
| ) |
There was a problem hiding this comment.
The non-greedy brace matching \{.*?\} in the _TOOL_USE_RE regular expression will stop at the first closing brace }. If the JSON payload contains nested objects, this regex will truncate the JSON string, causing json.loads to fail with a JSONDecodeError.\n\nSince the JSON is enclosed by the </tool_use> tag, we can safely match any characters up to the closing tag using (.*?) instead.
_TOOL_USE_RE = re.compile(\n r'<tool_use\\s+name="([^"]+)"[^>]*?(?:/\\s*>|>\\s*(.*?)\\s*</tool_use>)', re.DOTALL\n)
Comment on lines
+155
to
+160
| result = await db.execute( | ||
| select(TaskRunEvent) | ||
| .where(TaskRunEvent.run_id == run_id) | ||
| .order_by(TaskRunEvent.created_at, TaskRunEvent.id) | ||
| ) | ||
| events = result.scalars().all() |
There was a problem hiding this comment.
Polling the entire history of TaskRunEvent records for the active run every second is highly inefficient and does not scale. As the run progresses and accumulates events, the query will fetch and serialize an increasing number of rows on every single poll, only to discard the already-seen ones in Python.\n\nTo optimize this, you should track the last seen event's timestamp or ID and only query for new events (e.g., where(TaskRunEvent.run_id == run_id, TaskRunEvent.created_at > last_seen_time)).
Comment on lines
+54
to
+57
| if self.require_odp: | ||
| # Dual-write required: surface the failure even though legacy wrote. | ||
| logger.error("odp forward failed under require_odp: %s", exc) | ||
| raise |
There was a problem hiding this comment.
When require_odp is enabled, raising an exception here on ODP failure will cause the entire pipeline run to be marked as failed (success=False in run_pipeline), which completely skips downstream AI processing and notification steps.\n\nHowever, the legacy database write (self.legacy.write_batch) has already succeeded and committed the data. This creates a severe coupling where a transient ODP failure blocks all legacy downstream processing for successfully persisted records, and retrying the task might lead to duplicate processing or unique constraint violations. Consider decoupling the ODP write failure from the legacy pipeline completion, or handling this state transition more gracefully.
Comment on lines
+58
to
+66
| count_stmt = select(Skill) | ||
| if domain is not None: | ||
| stmt = stmt.where(Skill.domain == domain) | ||
| count_stmt = count_stmt.where(Skill.domain == domain) | ||
| if enabled is not None: | ||
| stmt = stmt.where(Skill.enabled.is_(enabled)) | ||
| count_stmt = count_stmt.where(Skill.enabled.is_(enabled)) | ||
|
|
||
| total = len((await db.execute(count_stmt)).scalars().all()) |
There was a problem hiding this comment.
Using len(scalars().all()) to get the total count of skills is highly inefficient because it loads all matching Skill records from the database into memory. If the database grows large, this will cause significant memory overhead and latency.\n\nInstead, use SQLAlchemy's func.count() to perform the count query directly on the database side.
from sqlalchemy import func\n count_stmt = select(func.count()).select_from(Skill)\n if domain is not None:\n stmt = stmt.where(Skill.domain == domain)\n count_stmt = count_stmt.where(Skill.domain == domain)\n if enabled is not None:\n stmt = stmt.where(Skill.enabled.is_(enabled))\n count_stmt = count_stmt.where(Skill.enabled.is_(enabled))\n\n total = (await db.execute(count_stmt)).scalar() or 0| const rect = el.getBoundingClientRect(); | ||
| if (rect.width === 0 && rect.height === 0) continue; | ||
|
|
||
| el.setAttribute('data-skill-ref', String(ref)); |
There was a problem hiding this comment.
Injecting custom data-skill-ref attributes directly into the live DOM is simple and clean, but on highly dynamic Single Page Applications (SPAs) built with React, Vue, or Svelte, virtual DOM re-renders can easily strip these custom attributes between the perception snapshot and the action execution.\n\nIf the attributes are stripped, subsequent selectors like [data-skill-ref="N"] will fail to find the elements. Consider adding a fallback mechanism or warning the operator about potential flakiness on highly dynamic SPA pages.
| role = str(el.get("role", "") or "") | ||
| name = str(el.get("name", "") or "") | ||
| value = str(el.get("value", "") or "") | ||
| line = f'#{ref} {role} "{name}"' |
There was a problem hiding this comment.
If the accessible name of an element contains double quotes (e.g., Click "Submit" button), this line will produce unescaped nested double quotes in the prompt (e.g., #3 button "Click "Submit" button"). This can confuse the LLM when parsing the element list.\n\nConsider escaping or stripping double quotes from the name before rendering.
escaped_name = name.replace('"', '\\"')\n line = f'#{ref} {role} "{escaped_name}"'
Comment on lines
+101
to
+106
| if isinstance(traces, dict): | ||
| trace = traces | ||
| else: | ||
| if not traces: | ||
| raise ValueError("re_distill requires at least one trace") | ||
| trace = traces[-1] # v1: distill the most recent failing trace |
There was a problem hiding this comment.
There is no type validation on traces when it is not a dictionary. If traces is passed as a string or a list of strings/integers instead of dictionaries, trace = traces[-1] will resolve to a string/scalar, and the subsequent call to distill_trace(trace, provider) will raise an AttributeError when trying to call .get() on it.\n\nEnsure that trace is validated to be a dictionary before proceeding.
if isinstance(traces, dict):\n trace = traces\n else:\n if not traces:\n raise ValueError("re_distill requires at least one trace")\n trace = traces[-1]\n if not isinstance(trace, dict):\n raise ValueError("trace must be a dictionary")…ggers edge label node --test on .ts files needs Node's built-in type-stripping (default-on since 23.6), unavailable on the pinned Node 20 runner -> ERR_UNKNOWN_FILE_EXTENSION for every frontend test file. topologyModel.ts labeled the manual-trigger source->task edge as "manual" instead of "triggers", so buildTopologyGraph never emitted the edge the test (and downstream consumers) expect.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
thin-channel-thick-runner → main
Lands the long-running
refactor/thin-channel-thick-runnerbranch. Headline work is a strangler-fig refactor of the collection pipeline ("thin channel + thick runner": adding a data source ≈ 100 lines) plus Phase 2/3 hardening. The branch also carries earlier work already developed on it (skills dock, node-kit, topology canvas).Strangler-fig channel refactor (PR1–PR5b)
LegacyDbSinkwrite seam — pipeline writes through anItemSink, behavior unchanged.RecordEvent/OdpIngestResponsemirroring the Rustodp-contracts;triple_to_eventroutes through one mapper.OdpSink;DualSinkwrites legacy + ODP exactly once (no double-send) via aforward_to_odpgate.data_sources.write_strategystate machine selects the sink (legacy / odp_shadow / odp_dual_required / odp_primary / odp_only).source_cursorstable +DBCursorStore; RSS on the thick contract (fetch()etag/304 conditional GET,identity()= entry id).run_channel; the cursor is committed only after the write sink durably accepts the batch.Phase 2/3
capabilities.session_affinityinstead of a hardcoded channel list.AuthManager+ Fernet-encryptedsource_credentials— secrets leavechannel_configplaintext; channels get a decryptedAuthContext. Inline-plaintext auth is deprecated with a warning.PER_DOMAIN_CONCURRENCY, default 3).Notes
tests/unit).o5j6k7l8m9n0write_strategy,p6k7l8m9n0o1source_cursors,q7l8m9n0o1p2source_credentials) — runalembic upgrade head.CREDENTIAL_ENCRYPTION_KEY(Fernet key) in prod for the credential store.api_channelontofetch()+AuthManager; swap the per-domain limiter to Redis for a multi-worker fleet.