[examples] Bluesky/ATProto sample integration (input stream, rules) - #236
Merged
Conversation
Mirrors the existing example_plugins/example_rules pattern with a custom register_input_stream hook that subscribes to Bluesky's JetStream WebSocket firehose. Lets contributors exercise Osprey against real production-rate event volume without the synthetic 1-event/sec generator. Stack docker-compose.atproto.yaml on top of the main compose file (or run ./run-atproto.sh) to use it. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
![github-code-quality[bot]](https://avatars.githubusercontent.com/u/9919?s=60&v=4){kind=small}
haileyok
commented
Apr 30, 2026
- Action data shape now mirrors haileyok/atproto-ruleset: $.did,
$.operation.{action,collection,path,cid,record}, $.eventMetadata.
Identity events become action_name='identity'; commit events become
'operation#<create|update|delete>'.
- Rules tree restructured to match the index.sml routing pattern:
main.sml → rules/index.sml → rules/record/index.sml →
rules/record/post/index.sml → individual rule files. Models split
into models/{base,record/base,record/post}.sml.
- Use required=False on entities so non-applicable events don't emit
extraction errors (eliminates __error_count noise on non-post events).
- Snowflake-generated action_id via batched generate_snowflake_batch
(250 IDs per call) instead of time_us, since collisions are possible
at sustained throughput.
- Add unit tests for _event_to_action covering posts, likes, deletes,
identity, account skip, missing time_us, unknown kinds.
- Add app.bsky.actor.profile to default subscribed collections.
- Tighten per-event try/except so a malformed event no longer tears
down the WebSocket connection.
- Drop the Discord-monorepo-specific port-reset block from
docker-compose.atproto.yaml.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
mypy couldn't narrow `ws: Optional[WebSocket]` inside the recv loop because the variable also had to survive into the `finally` block. Moved the inner streaming + close lifecycle into _stream_one_connection, leaving _gen as just the reconnect loop. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Earlier review feedback pointed to haileyok/atproto-ruleset for project
*structure* — but that ruleset is fed by a separate enrichment pipeline,
not JetStream directly, and uses paths like \$.eventMetadata.* and
\$.operation.path that JetStream doesn't emit. Mapping JetStream events
into that shape was misleading: it pretended to expose enrichment
fields that are always null and synthesised paths that don't exist.
Reverted the data shape so Action.data is the JetStream JSON event
passed through unchanged. Rules now read JetStream-native paths:
\$.did, \$.kind, \$.commit.{operation,collection,rkey,cid,record},
\$.identity.handle, etc. action_name is the event's kind ('commit'
or 'identity'). The file hierarchy from atproto-ruleset (main.sml,
models/{base,record/...}, rules/index.sml routing) is preserved.
Verified end-to-end on the live worker:
- 7/7 unit tests pass in container
- 230,392 events processed post-restart, every one with __error_count: 0
- 253,820 commits + 142 identity events emitted with correct paths
- 70 PostContainsTestRule positive matches with label mutation
UserId/test-poster/1 emitted on posts containing "test"
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
haileyok
changed the title
- IMP-1: Add 30s socket read timeout to detect stalled connections - IMP-2: Track in-process cursor (time_us) to resume from last event on reconnect - IMP-3: Add test coverage for _build_url wantedCollections format, cursor param presence, and malformed message handling - IMP-4: Drop snowflake-id-worker dependency; pass action_id=0 to rely on RulesSink fallback minting - MIN-5: Drop coerce_type=True from string-typed JsonData fields (Collection, Rkey, Cid, PostText) - MIN-7: Stricter time_us validation: reject 0 and negative values, add unit tests - MIN-8: Change debug log to info for socket-close errors - MIN-9: Clarify docker-compose stack startup time in README - MIN-6: Update atproto-ruleset caveat to note enrichment pipeline shape mismatch Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
The new test from the previous commit never actually ran (pytest can't import the workspace package outside Docker). Inside Docker the test revealed two issues: - WebSocketConnectionClosedException propagated out of _stream_one_connection, breaking list(gen). Catch it inside the loop and return cleanly — a closed connection is the expected end of a session, not an error worth reraising. _gen still reconnects since the generator just finishes. - NoopAckingContext exposes _item, not item; tests now use the actual attribute name. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
JetStream's contract is time_us: int (microseconds). The previous isinstance(time_us, (int, float)) was over-permissive and would have allowed _event_to_action to yield while the cursor-update narrowing at the call site (isinstance(time_us, int)) silently rejected the same value — leaving the cursor stale on reconnect. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
The two earlier commits on this branch that touched uv.lock (469219b, c2c0746) were generated with the older host uv, which downgraded the lockfile from rev 3 (main's format, with editable workspace sources) to rev 2 (with virtual workspace sources). Docker builds use a newer uv that requires rev 3, so `uv sync --locked` was failing in the worker image build. Regenerated with the uv shipped in the worker image so the lockfile again matches main's format. Source-only changes: revision bump, workspace members marked editable, and two unused linux_armv7l grpcio wheels dropped. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Plugin emits action_name = '<operation>_<short>' for commit events (create_post, delete_like, update_profile, ...) using the COLLECTION_NAMES map next to DEFAULT_COLLECTIONS, or 'identity' for identity events. Commits for unmapped collections or unexpected operations are skipped. Rules side adds three small JsonData feature models — IdentityHandle, LikeSubjectUri, FollowSubject — and imports them from main.sml so they evaluate for every action (None when the path doesn't resolve). example_atproto_rules/config/ui_config.yaml registers default_summary_features mapping action globs to the features the Osprey UI surfaces in the event stream — so each event type shows contextually relevant fields without rule-code changes. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
The cycle-1 IMP-4 review change passed action_id=0 expecting RulesSink.run's fallback to mint a snowflake. The fallback's condition is `if not action.action_id and action.action_id != 0`, which short- circuits to False when action_id is exactly 0 — so action_id stays 0 forever. Every event landed in MinIO storage keyed action_id=0, overwriting the previous one. Druid retained N distinct rows but the event-scan UI fetched by action_id, returning the same MinIO object N times — hence the duplicate-event rendering. Restoring the local snowflake batch (250 ids per snowflake-id-worker call) so each event gets a distinct id. README updated to surface the SNOWFLAKE_API_ENDPOINT dependency. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Old shape evaluated FollowSubject ($.commit.record.subject as str) for
every event. For like/repost records, the path resolves to a {uri, cid}
dict, the str type check fails, and the engine increments
__error_count by 1 on every like/repost — likely also what was
breaking the timeseries chart, since errored events get aggregated
differently.
Replace with a single Subject feature in models/record/base.sml that
prefers $.commit.record.subject.uri (the dict-shape) and falls back to
$.commit.record.subject coerced to str (the follow DID-string shape)
via ResolveOptional. Drop the per-collection follow.sml / like.sml
files and update ui_config.yaml so likes / reposts / follows all
reference the unified Subject.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Member
Author
|
haileyok
marked this pull request as ready for review
May 7, 2026 23:09
haileyok
requested review from
a team,
EXBreder,
ayubun and
vinaysrao1
as code owners
May 7, 2026 23:09
haileyok
commented
May 8, 2026
| SNOWFLAKE_BATCH_SIZE = 250 | ||
|
|
||
|
|
||
| class JetStreamInputStream(BaseInputStream[BaseAckingContext[Action]]): |
Member
Author
There was a problem hiding this comment.
this is the main thing i'd love to get eyes on if anyone has python websocket experience...it doesn't need to be production stable since its really a testing/example guy (not to mention jetstream shouldn't be used for real-world moderation tasks anyway) but it should be at least mostly stable...
Member
Author
There was a problem hiding this comment.
i did want to opt for a non-kafka option in here, because doing this helps break the misconception that osprey can only be used with kafka
juanmrad
reviewed
May 8, 2026
juanmrad
left a comment
juanmrad
left a comment
Member
There was a problem hiding this comment.
Left some comments. My main concern is using WebSocket over WebSocketApp. we'd get the connection lifecycle (ping_interval / pong handling) for free.
|
|
||
| def _stream_one_connection(self, url: str) -> Iterator[BaseAckingContext[Action]]: | ||
| logger.info(f'Connecting to JetStream at {url}') | ||
| ws = websocket.create_connection(url, timeout=30) |
Member
There was a problem hiding this comment.
websocket.create_connection() returns the low-level WebSocket instance, which doesn't auto-PING. ref
Meaning there is no ping interval ensuring the connection stays alive. So today the only thing keeping the connection healthy is JetStream's volume + the read timeout. Here you can either spawn a daemon to health ping the socket.
Or better yet use WebSocketApp.run_forever(ping_interval=20, ping_timeout=10) and bridge the callback API into the _gen iterator via a gevent.queue.Queue. ref
docs explain my thought here.
Comment on lines
+65
to
+66
| def _gen(self) -> Iterator[BaseAckingContext[Action]]: | ||
| while True: |
Member
There was a problem hiding this comment.
We should provably want to add a backoff. If JetStream is down or rate-limiting, you reconnect every 2 seconds in a tight loop and trigger escalating errors, potential higher rate limits and sentry capture on every event. We can start with a simple
backoff = self._reconnect_seconds
while True:
.....
backoff = min(backoff * 2, 60.0)
Comment on lines
+91
to
+94
| except Exception: | ||
| logger.exception('skipping malformed JetStream event') | ||
| sentry_sdk.capture_exception() | ||
| continue |
Member
There was a problem hiding this comment.
Suggested change
| except Exception: | |
| logger.exception('skipping malformed JetStream event') | |
| sentry_sdk.capture_exception() | |
| continue | |
| except json.JSONDecodeError: | |
| logger.warning('skipping malformed JetStream JSON') | |
| continue |
we should be more specific on the error.
websocket.create_connection returns the low-level socket which doesn't auto-PING; the only thing keeping the connection alive was JetStream's volume plus the read timeout. Use WebSocketApp.run_forever with ping_interval=20 and ping_timeout=10 instead, and bridge the callback API into _gen via a gevent.queue.Queue. greenlet.link pushes a 'done' sentinel as a safety net so the generator never blocks if run_forever exits without firing on_close. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Previously every reconnect waited a fixed reconnect_seconds (default 2s), so a dead JetStream host or a bad URL would just be hammered every two seconds forever. Add max_reconnect_seconds (default 60s) and double the sleep on each session that produced no events; reset to the base on any session that yielded at least one event. Extracted the state update into _advance_backoff so it can be tested without spinning up the full _gen loop. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
julietshen
added a commit
that referenced
this pull request
Jun 1, 2026
Threaded Kafka consumer that subscribes to the execution-results topic and records the wall-clock receive time per ActionId. Producer-agnostic: * `action_id_filter` set → closed-loop matching (record only known IDs) * `action_id_filter=None` → open-loop counting (record everything) The open-loop mode is what will let the same code measure jetstream traffic once #236 lands — same module, different config. Belt-and-suspenders on the parsing side: skips malformed JSON, missing ActionId, non-integer ActionId. First-write-wins for duplicate IDs so duplicate effect dispatches don't reset the receive timestamp. Bounded poll cycle (`consumer_timeout_ms=200`) lets the stop signal fire within ~200ms regardless of message arrival rate. Factory-injected KafkaConsumer so unit tests run without live Kafka. Part of the stress harness work for #324, split into smaller PRs per AGENTS.md. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
14 tasks
julietshen
added a commit
that referenced
this pull request
Jun 16, 2026
Threaded Kafka consumer that subscribes to the execution-results topic and records the wall-clock receive time per ActionId. Producer-agnostic: * `action_id_filter` set → closed-loop matching (record only known IDs) * `action_id_filter=None` → open-loop counting (record everything) The open-loop mode is what will let the same code measure jetstream traffic once #236 lands — same module, different config. Belt-and-suspenders on the parsing side: skips malformed JSON, missing ActionId, non-integer ActionId. First-write-wins for duplicate IDs so duplicate effect dispatches don't reset the receive timestamp. Bounded poll cycle (`consumer_timeout_ms=200`) lets the stop signal fire within ~200ms regardless of message arrival rate. Factory-injected KafkaConsumer so unit tests run without live Kafka. Part of the stress harness work for #324, split into smaller PRs per AGENTS.md. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
5 tasks
julietshen
added a commit
that referenced
this pull request
Jun 17, 2026
Orchestrates the stress reporter (#328), producer (#329), and consumer (#330) into a CLI that produces synthetic events at a configurable rate, observes their ExecutionResults, and reports drop rate + p50/p95/p99 latency. Exits non-zero on threshold breach so it can gate CI on pipeline health. Subcommands: * `run` — closed-loop synthetic testing. Blocked on #330; the body lazy-imports the consumer and short-circuits with a clear pointer message until that PR merges, then activates. * `measure` — open-loop measurement against an external source. Stub; will activate once #236 (jetstream input stream plugin) lands. Argparse + threshold gates + report dispatch + entry-point registration are all live today, so the CLI structure (~240 lines) is reviewable independently. When #330 merges, the only change needed here is dropping the `try/except ImportError` wrapper around the consumer import — the orchestration body below it already references Consumer/ConsumerConfig. Adds: * `osprey-stress` console script in `osprey_worker/pyproject.toml` * CHANGELOG entry citing the full stack of PRs the harness landed across * 7 unit tests covering arg parsing, the measure stub, and the #330-blocked-on-import path so a regression in either is loud
julietshen
added a commit
that referenced
this pull request
Jun 17, 2026
Orchestrates the stress reporter (#328), producer (#329), and consumer (#330) into a CLI that produces synthetic events at a configurable rate, observes their ExecutionResults, and reports drop rate + p50/p95/p99 latency. Exits non-zero on threshold breach so it can gate CI on pipeline health. Subcommands: * `run` — closed-loop synthetic testing. Blocked on #330; the body lazy-imports the consumer and short-circuits with a clear pointer message until that PR merges, then activates. * `measure` — open-loop measurement against an external source. Stub; will activate once #236 (jetstream input stream plugin) lands. Argparse + threshold gates + report dispatch + entry-point registration are all live today, so the CLI structure (~240 lines) is reviewable independently. When #330 merges, the only change needed here is dropping the `try/except ImportError` wrapper around the consumer import — the orchestration body below it already references Consumer/ConsumerConfig. Adds: * `osprey-stress` console script in `osprey_worker/pyproject.toml` * CHANGELOG entry citing the full stack of PRs the harness landed across * 7 unit tests covering arg parsing, the measure stub, and the #330-blocked-on-import path so a regression in either is loud
chimosky
reviewed
Jun 18, 2026
| return None | ||
| commit = event.get('commit') or {} | ||
| operation = commit.get('operation') | ||
| collection = commit.get('collection', '') or '' |
Contributor
There was a problem hiding this comment.
I guess this wasn't intentional, as there's a default return value here so the or is redundant.
julietshen
added a commit
that referenced
this pull request
Jun 30, 2026
Orchestrates the stress reporter (#328), producer (#329), and consumer (#330) into a CLI that produces synthetic events at a configurable rate, observes their ExecutionResults, and reports drop rate + p50/p95/p99 latency. Exits non-zero on threshold breach so it can gate CI on pipeline health. Subcommands: * `run` — closed-loop synthetic testing. Blocked on #330; the body lazy-imports the consumer and short-circuits with a clear pointer message until that PR merges, then activates. * `measure` — open-loop measurement against an external source. Stub; will activate once #236 (jetstream input stream plugin) lands. Argparse + threshold gates + report dispatch + entry-point registration are all live today, so the CLI structure (~240 lines) is reviewable independently. When #330 merges, the only change needed here is dropping the `try/except ImportError` wrapper around the consumer import — the orchestration body below it already references Consumer/ConsumerConfig. Adds: * `osprey-stress` console script in `osprey_worker/pyproject.toml` * CHANGELOG entry citing the full stack of PRs the harness landed across * 7 unit tests covering arg parsing, the measure stub, and the #330-blocked-on-import path so a regression in either is loud
Resolve conflicts: - pyproject.toml: keep both osprey_async_worker (main) and example_atproto_plugins (this branch) across workspace members, isort, and fawltydeps. - CHANGELOG.md: adopt main's Keep a Changelog format; re-add the #236 entry under Added. - uv.lock: regenerated with `uv lock` (picks up websocket-client and example-atproto-plugins).
The atproto plugin ships 24 unit tests for the JetStream event mapping, but example_atproto_plugins was not in testpaths, so pytest never collected them. Add it alongside example_plugins. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01TEcfpqhza3dPepZUVWw36X
Contributor
|
Warning Review limit reached
Next review available in: 27 minutes Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. Review details⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (5)
📝 WalkthroughWalkthroughAdds an ATProto JetStream WebSocket input plugin, example rules and UI configuration, Docker Compose integration, workspace setup, documentation, and tests for converting JetStream events into Osprey actions. ChangesATProto JetStream integration
Estimated code review effort: 3 (Moderate) | ~25 minutes Sequence Diagram(s)sequenceDiagram
participant JetStream
participant JetStreamInputStream
participant OspreyAction
participant ExampleRules
JetStream->>JetStreamInputStream: Send event over WebSocket
JetStreamInputStream->>OspreyAction: Yield mapped action
OspreyAction->>ExampleRules: Evaluate action fields
ExampleRules->>OspreyAction: Apply test-poster label when matched
Suggested reviewers: 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
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 |
The atproto plugin registers only the input stream; the sample rules use TextContains, a labels service, and an output sink from the sibling example_plugins package. Document that the two run together so it isn't a surprise for anyone lifting the sample on its own. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01TEcfpqhza3dPepZUVWw36X
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 3
🧹 Nitpick comments (2)
docker-compose.atproto.yaml (1)
11-12: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueUse a consistent absolute path for
OSPREY_RULES_PATHacross services.The worker uses a relative path (
./example_atproto_rules) while the UI API uses an absolute path (/osprey/example_atproto_rules). Both resolve correctly because the worker'sWORKDIRis/osprey, but the inconsistency is fragile — a futureWORKDIRchange would silently break rule loading. Prefer the absolute path in both services.♻️ Proposed fix
services: osprey-worker: environment: OSPREY_INPUT_STREAM_SOURCE: plugin - OSPREY_RULES_PATH: ./example_atproto_rules + OSPREY_RULES_PATH: /osprey/example_atproto_rules volumes:Also applies to: 19-19
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@docker-compose.atproto.yaml` around lines 11 - 12, Update the OSPREY_RULES_PATH environment variable in the worker service to use the same absolute /osprey/example_atproto_rules path as the UI API service, keeping rule loading consistent across services.example_atproto_plugins/src/atproto_plugin/jetstream_input_stream.py (1)
93-93: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winUnbounded queue can grow without limit if the consumer falls behind.
queue = Queue()has nomaxsize.on_messagepushes every incoming JetStream frame into it as fast as the server sends; if the generator's consumer (downstream Osprey action processing) is slower than the firehose, the queue has nothing bounding its growth — a legitimate memory-growth risk against a live, high-throughput source.Consider giving the queue a bounded
maxsizesoqueue.put()in the websocket greenlet naturally applies backpressure (note: this will also delay ping/pong handling on that greenlet if the consumer is very slow, so pick a size that balances buffering vs. keepalive risk).Also applies to: 98-99, 124-156
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@example_atproto_plugins/src/atproto_plugin/jetstream_input_stream.py` at line 93, Bound the queue used by the JetStream input stream to prevent unbounded memory growth when consumption lags behind incoming frames. Update the Queue initialization near on_message and ensure its maxsize is large enough to provide buffering while allowing websocket keepalive handling to remain responsive; preserve the existing queue.put and generator consumption behavior.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@example_atproto_plugins/pyproject.toml`:
- Around line 6-9: Update the dependencies list to flag websocket-client==1.8.0
for required approval before merge, ensuring the new dependency undergoes
license and CVE review while leaving the already-covered pluggy==1.5.0 entry
unchanged.
In `@example_atproto_plugins/src/atproto_plugin/jetstream_input_stream.py`:
- Around line 57-61: Prevent repeated snowflake mint attempts from occurring for
every queued message during an outage. Update _next_action_id and its caller in
_stream_one_connection so generate_snowflake_batch failures propagate or
otherwise terminate the current connection loop, allowing _gen()'s existing
backoff/reconnect delay to throttle subsequent retries; preserve normal
buffering and action-ID generation on successful batches.
- Around line 170-172: Bound-validate time_us in the event parsing logic before
converting it with datetime.fromtimestamp, including the corresponding
validation at the alternate occurrence around line 186. Reject values whose
converted timestamp would be outside the supported datetime range so malformed
events return None and do not propagate an exception into the connection
generator.
---
Nitpick comments:
In `@docker-compose.atproto.yaml`:
- Around line 11-12: Update the OSPREY_RULES_PATH environment variable in the
worker service to use the same absolute /osprey/example_atproto_rules path as
the UI API service, keeping rule loading consistent across services.
In `@example_atproto_plugins/src/atproto_plugin/jetstream_input_stream.py`:
- Line 93: Bound the queue used by the JetStream input stream to prevent
unbounded memory growth when consumption lags behind incoming frames. Update the
Queue initialization near on_message and ensure its maxsize is large enough to
provide buffering while allowing websocket keepalive handling to remain
responsive; preserve the existing queue.put and generator consumption behavior.
🪄 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: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: 79af949e-18e5-4859-9c35-76089f76dc93
⛔ Files ignored due to path filters (1)
uv.lockis excluded by!**/*.lock
📒 Files selected for processing (25)
AGENTS.mdCHANGELOG.mddocker-compose.atproto.yamlexample_atproto_plugins/README.mdexample_atproto_plugins/__init__.pyexample_atproto_plugins/pyproject.tomlexample_atproto_plugins/src/atproto_plugin/__init__.pyexample_atproto_plugins/src/atproto_plugin/jetstream_input_stream.pyexample_atproto_plugins/src/atproto_plugin/register_plugins.pyexample_atproto_plugins/tests/__init__.pyexample_atproto_plugins/tests/test_jetstream_input_stream.pyexample_atproto_rules/config/labels.yamlexample_atproto_rules/config/ui_config.yamlexample_atproto_rules/main.smlexample_atproto_rules/models/base.smlexample_atproto_rules/models/identity.smlexample_atproto_rules/models/record/base.smlexample_atproto_rules/models/record/post.smlexample_atproto_rules/rules/index.smlexample_atproto_rules/rules/record/index.smlexample_atproto_rules/rules/record/post/index.smlexample_atproto_rules/rules/record/post/post_contains_test.smlosprey_worker/Dockerfilepyproject.tomlrun-atproto.sh
Comment on lines
+6
to
+9
| dependencies = [ | ||
| "pluggy==1.5.0", | ||
| "websocket-client==1.8.0", | ||
| ] |
Contributor
There was a problem hiding this comment.
🔒 Security & Privacy | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Verify websocket-client license and check for known CVEs
gh api graphql -f query='
{
securityVulnerabilities(first: 10, ecosystem: PIP, package: "websocket-client") {
nodes {
advisory { summary severity publishedAt }
vulnerableVersionRange
firstPatchedVersion { identifier }
}
}
}'Repository: roostorg/osprey
Length of output: 202
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect the relevant pyproject files and find all websocket-client mentions.
git ls-files '**/pyproject.toml' | sort
printf '\n--- root pyproject excerpt ---\n'
sed -n '1,120p' pyproject.toml
printf '\n--- package pyproject excerpt ---\n'
sed -n '1,120p' example_atproto_plugins/pyproject.toml
printf '\n--- websocket-client references ---\n'
rg -n --hidden --glob '!*node_modules*' --glob '!*dist*' --glob '!*build*' 'websocket-client' .Repository: roostorg/osprey
Length of output: 4819
Require approval for websocket-client==1.8.0 before merge. It’s a new dependency and needs license/CVE review; pluggy==1.5.0 is already covered by the shared deps.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@example_atproto_plugins/pyproject.toml` around lines 6 - 9, Update the
dependencies list to flag websocket-client==1.8.0 for required approval before
merge, ensuring the new dependency undergoes license and CVE review while
leaving the already-covered pluggy==1.5.0 entry unchanged.
Source: Path instructions
…check The license check (added since this branch was cut) flags the local example_atproto_plugins package as UNKNOWN license. Add it to the first-party ignore list alongside example_plugins, and trigger the check on its pyproject. websocket-client (Apache-2.0) is already covered by the allow-list. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01TEcfpqhza3dPepZUVWw36X
python-quality (added since this branch was cut) flagged two things: the atproto_plugin package lacked a py.typed marker, so mypy treated imports from it as untyped; and jetstream_input_stream.py needed reformatting under main's ruff line length. Add the marker and apply ruff format. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01TEcfpqhza3dPepZUVWw36X
- Snowflake mint failures now drop the connection instead of re-minting per message, so _gen()'s reconnect backoff throttles retries during a snowflake-id-worker outage (was a retry-storm risk). - Guard datetime.fromtimestamp against out-of-range time_us so one malformed event is skipped rather than propagating an exception that forces a full reconnect; add a regression test. - Use an absolute OSPREY_RULES_PATH for the worker to match the UI API. websocket-client (Apache-2.0) already passes the license check and has no known CVEs, so no dependency change was needed. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01TEcfpqhza3dPepZUVWw36X
julietshen
self-requested a review
July 13, 2026 18:51
julietshen
approved these changes
Jul 13, 2026
julietshen
left a comment
julietshen
left a comment
Member
There was a problem hiding this comment.
finished this PR by cleaning up the conflicts and updated changelog, tested this locally and jetstream is populating
This was referenced Jul 13, 2026
julietshen
added a commit
that referenced
this pull request
Jul 13, 2026
The enrichment does a per-event external API call, which is great for demos but undermines #236's load-testing purpose. Move Handle/DisplayName out of base.sml into an opt-in models/enrichment.sml that main.sml does not import by default, so the firehose runs dependency-free unless enrichment is explicitly enabled. Keep the two shipped UDFs (AtprotoHandle, AtprotoDisplayName) over a shared full-profile cache, and document in the README both how to enable enrichment and how to extend it with more profile fields (account age, follower counts, labels) rather than shipping them all wired-in. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01TEcfpqhza3dPepZUVWw36X
julietshen
added a commit
that referenced
this pull request
Jul 20, 2026
Cross-checked the docs against the engine and CLI and fixed several things that were wrong or would not actually work: - cli-reference: osprey-cli subcommands use dashes, not underscores (Click derives the command name from the function and converts _ to -), so `push_rules` etc. failed with "No such command". Also refreshed the osprey-stress `measure` wording now that #236 has merged. - query-syntax: SML uses `None`, not `Null`; the query box supports only four built-in functions (RegexMatch, DidAddLabel, DidRemoveLabel, DidDeclareVerdict), so the TextContains/ListLength query examples (both rules UDFs) would 500. Reworked the "Using UDFs" section and the note. - manage: clarified that the UDF Registry is mainly a rules reference; only those four functions work in the query box. - rules/README and rules/examples: Rule(...) requires a `description` argument (validation rejects it otherwise) and the null literal is `None`, not `Null`. Verified both against the real validator (validate_and_push). - local: use `uv run python` instead of a hardcoded python3.11 binary. Co-Authored-By: cassidyjames <611168+cassidyjames@users.noreply.github.com> Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
julietshen
added a commit
that referenced
this pull request
Jul 22, 2026
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
5 participants
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.
Description
Adds a new workspace package,
example_atproto_plugins, demonstrating how to plug Osprey into a real-world event source via the existingregister_input_streampluggy hook. The sample subscribes to Bluesky's JetStream WebSocket firehose and maps each ATProto commit (posts, likes, reposts, follows) into an OspreyAction. Companion rules live underexample_atproto_rules/.The main motivation is making it cheap to exercise Osprey changes against realistic event shapes and volume — the existing
osprey-kafka-test-data-produceremits one synthetic event per second from a single template, which doesn't catch issues that only show up with real production traffic. JetStream is free, public, no-auth, and runs at hundreds of events per second; in a few minutes of local testing this stack processed ~34k events of varied content.A secondary benefit: it fills the gap between the README's "used by Bluesky" claim and the lack of any actual ATProto reference code in this repo. The existing
example_pluginsis Discord-shaped; this is the Bluesky-shaped counterpart.Usage
Stacks
docker-compose.atproto.yamlon top of the main compose file, switching the worker fromOSPREY_INPUT_STREAM_SOURCE=kafkato=pluginand pointing it atexample_atproto_rules/. The existing demo path (./demo.sh) is unchanged.Notes / non-goals
websocket-client==1.8.0(Apache-2.0).uv.lockrevision bumps from 2 to 3 because the project's[build-system]requiresuv_build>=0.8.12,<0.9.0, which expects the newer lockfile format.osprey_worker/Dockerfilepicks up the new workspace member alongsideexample_plugins.Confidence Level
Confidence Level: Claude
Testing
_event_to_actionmapping for posts, likes, deletes, identity-event skip, and missingtime_us.docker compose -f docker-compose.yaml -f docker-compose.atproto.yaml configmerges cleanly; maindocker-compose.yamlstill validates standalone.uv run ruff check,uv run ruff format --checkclean.register_plugins(existing UDFs) andatproto_plugins(new input stream); no name collision.PostContainsTestrule evaluates correctly against real post text (gated byEventType == 'create_post').Checklist
uv run ruff check .passesuv tool run fawltydeps --check-unused --pyenv .venv— not run on host (venv permission issue from a docker volume mount); CI will check.CHANGELOG.md— leaving for reviewer to decide on appropriate entry.Summary by CodeRabbit