feat(events): NATS publish plumbing + a versioned prompt registry - #394
Conversation
Phase 1 of the case-enrichment event bus, in the two parts that are pure
code and merge safely with no broker running. Nothing here changes the
Case write path, and there is no migration: `events` has no models and
`Job.kind` is already free-form.
events/ — the bus
Envelope, subject vocabulary, JetStream stream topology, and a
sync->async publisher. Three properties it exists to guarantee:
- Publishing is best-effort and NEVER raises. A broker outage must
degrade enrichment, not a case write; the bus is transport and the
case record is the system of record.
- One connection per process, not one per publish. A background loop
thread holds a long-lived connection, started lazily so it survives
a gunicorn fork. Connecting per call would turn a burst of approvals
into a burst of handshakes, which this cluster has been bitten by.
- A dead broker must not slow the request path. After a failed connect
we fail fast for 30s before retrying, and publishes are
fire-and-forget with failures surfaced via a done-callback.
Streams are asserted from code rather than declared in YAML: NATS has
no CRD, `add_stream` is upsert-like, and this keeps the topology next
to the code that depends on it with no bootstrap step in a fresh
environment.
case_proposals/publish.py — the first real producer
Approve/reject emit jaw.case.update.{approved,rejected} via
transaction.on_commit, mirroring the existing best-effort
_schedule_reindex / _schedule_material_visibility hooks. Registered
inside the atomic block so a rollback discards it, fired only after a
successful commit so a subscriber can never read the case before the
write is visible. The decision's dedup_key is keyed on the decision
(proposal:<pk>:<status>), distinct from the fact's dedup_key, which is
carried in the payload.
llm/prompts.py — PromptSpec
A named, versioned record of (system, content builder, tier, max
tokens) that invoke_json is called with. Version matters because
"which prompt produced this wrong intent?" is the first question asked
of an LLM-generated record. Deliberately not a templating engine, and
migrating review/judge.py onto it is out of scope.
It validates `tier`, which is not decoration: routing.provider_for_tier
resolves "premium" and treats EVERYTHING else as cheap, so a typo would
silently downgrade the model instead of raising.
NATS_URL is unset by default: every publish is a logged no-op, no thread
is started, and dev/CI need no broker. Setting it is also the entire
rollback, with no image change.
Tests: 51 new. The load-bearing ones assert an approval still succeeds
and still applies its intent both with no broker configured and with the
publisher raising. Verified non-vacuous by mutation — removing the
publish call fails exactly the two tests that assert it.
Full suite: 2603 passed, 4 skipped.
Co-Authored-By: Claude <noreply@anthropic.com>
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
There was a problem hiding this comment.
Your trial has ended. Reactivate Greptile to resume code reviews.
|
Failed to generate code suggestions for PR |
PR Agent Walkthrough 🤖Welcome to the PR Agent, an AI-powered tool for automated pull request analysis, feedback, suggestions and more. Here is a list of tools you can use to interact with the PR Agent:
(1) Note that each tool can be triggered automatically when a new PR is opened, or called manually by commenting on a PR. (2) Tools marked with [*] require additional parameters to be passed. For example, to invoke the |
|
Auto-approved PR |
![jawafdehi-pr-agent[bot]](https://avatars.githubusercontent.com/u/241969530?s=60&v=4){kind=small}
|
Warning Review limit reached
Next review available in: 13 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: Organization UI Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (20)
📝 WalkthroughWalkthroughThe change adds an optional NATS/JetStream event bus, post-commit approval and rejection events, event envelope and stream definitions, and a versioned LLM prompt registry with validation, invocation, and registry tests. ChangesCase decision event bus
LLM prompt registry
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant DecisionView as case_proposals.views
participant Publisher as schedule_decision_event
participant Bus as events.bus.publish
participant JetStream
DecisionView->>Publisher: Save approved or rejected proposal
Publisher->>Publisher: Build decision envelope
Publisher->>Bus: Register on_commit callback
Bus->>JetStream: Publish serialized event
Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches 💡 1📝 Generate docstrings 💡
🧪 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 |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 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 `@events/bus.py`:
- Around line 108-143: Separate the NATS connection timeout from the overall
startup timeout in _start_locked and _connect: allow ensure_streams to complete
after nats.connect without the outer future expiring prematurely. If startup
still times out or fails, cancel the submitted _connect coroutine and ensure any
connection it creates is closed when cancellation or late completion occurs,
preventing a detached NATS connection from remaining alive.
In `@events/streams.py`:
- Around line 65-101: The ensure_streams function should tolerate expected
configuration drift: attempt add_stream first, and when it fails because the
stream already exists with conflicting configuration, fetch or reuse the
existing stream definition and call js.update_stream with the current
StreamConfig. Preserve propagation of genuine broker or setup failures, and keep
assertion logging and return behavior unchanged.
🪄 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: CHILL
Plan: Pro Plus
Run ID: bf348e23-5913-4f12-a93b-bb2f1be044fb
⛔ Files ignored due to path filters (1)
uv.lockis excluded by!**/*.lock
📒 Files selected for processing (16)
case_proposals/publish.pycase_proposals/tests/test_publish.pycase_proposals/views.pyconfig/settings.pyevents/__init__.pyevents/apps.pyevents/bus.pyevents/envelope.pyevents/streams.pyevents/subjects.pyevents/tests/__init__.pyevents/tests/test_bus.pyllm/prompts.pyllm/tests/__init__.pyllm/tests/test_prompts.pypyproject.toml
| def _start_locked(self): | ||
| loop = asyncio.new_event_loop() | ||
| thread = threading.Thread( | ||
| target=loop.run_forever, name="events-bus", daemon=True | ||
| ) | ||
| thread.start() | ||
|
|
||
| future = asyncio.run_coroutine_threadsafe(self._connect(), loop) | ||
| try: | ||
| self._nc, self._js = future.result(timeout=STARTUP_TIMEOUT_SECONDS) | ||
| except Exception: | ||
| loop.call_soon_threadsafe(loop.stop) | ||
| raise | ||
|
|
||
| self._loop = loop | ||
| self._thread = thread | ||
| self._pid = os.getpid() | ||
| logger.info("events.connected", url=_redact(settings.NATS_URL)) | ||
|
|
||
| async def _connect(self): | ||
| import nats | ||
|
|
||
| from events.streams import ensure_streams | ||
|
|
||
| nc = await nats.connect( | ||
| settings.NATS_URL, | ||
| name="jawafdehi-platform", | ||
| connect_timeout=STARTUP_TIMEOUT_SECONDS, | ||
| # Reconnect forever rather than giving up: this process outlives any | ||
| # broker restart, and a permanently-detached publisher that still | ||
| # looks healthy is worse than one that keeps trying. | ||
| max_reconnect_attempts=-1, | ||
| ) | ||
| js = nc.jetstream() | ||
| await ensure_streams(js) | ||
| return nc, js |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== locate files =="
fd -a 'bus.py|.*settings.*|.*dependencies.*|.*requirements.*|pyproject.toml|poetry.lock' . | sed 's#^\./##' | head -200
echo "== git status/stat =="
git diff --stat || true
echo "== events/bus.py outline and relevant lines =="
if [ -f events/bus.py ]; then
wc -l events/bus.py
sed -n '1,220p' events/bus.py | cat -n
fi
echo "== settings and connect timeout constants =="
rg -n "STARTUP_TIMEOUT_SECONDS|CONNECT_RETRY_SECONDS|NATS_URL|ensure_streams|_start_locked|events\.connect" -S .Repository: Jawafdehi/JawafdehiAPI
Length of output: 13451
🌐 Web query:
nats-py connect_timeout parameter source nats.connect connect_timeout
💡 Result:
In the nats-py library, the connect_timeout parameter is used within the nats.connect method to specify the maximum amount of time (in seconds) allowed to establish a connection to a NATS server [1][2][3]. Key details include: - Functionality: If multiple server URLs are provided, this timeout applies individually to the connection attempt for each server in the cluster [1][4]. - Default Value: The default value for connect_timeout is 2 seconds [2][3][5]. - Usage Example: import asyncio import nats async def main: nc = await nats.connect( servers=["nats://demo.nats.io:4222"], connect_timeout=10 # Set timeout to 10 seconds) await nc.close if name == 'main': asyncio.run(main) Note that while nats-py continues to support connect_timeout in its standard client, users should be aware of library migration paths (such as changes in newer core implementations) where parameter names can occasionally be updated or deprecated [6]. Always check the documentation for your specific version if you encounter unexpected behavior [2][7].
Citations:
- 1: https://docs.nats.io/using-nats/developer/connecting/connect_timeout
- 2: https://nats-io.github.io/nats.py/modules.html
- 3: https://github.com/nats-io/nats.py/blob/main/nats/src/nats/aio/client.py
- 4: https://github.com/nats-io/nats.docs/blob/master/using-nats/developing-with-nats/connecting/connect_timeout.md
- 5: https://github.com/nats-io/nats.py/blob/master/nats/aio/client.py
- 6: https://github.com/nats-io/nats.py/blob/e537164138a73f84cf67b4bfa624efa4092ee374/nats-core/MIGRATION.md
- 7: https://nats-io.github.io/nats.py/_modules/nats.html
🌐 Web query:
asyncio.run_coroutine_threadsafe timeout result future does not cancel task exception behavior
💡 Result:
When you call asyncio.run_coroutine_threadsafe(coro, loop), it returns a concurrent.futures.Future object that acts as a bridge to the asyncio Task running in the event loop [1][2]. If you wait for this future using future.result(timeout=...) and it times out, the following behavior applies: 1. Cancellation is not automatic: The TimeoutError raised by future.result only indicates that the thread waiting for the result has timed out [1][3]. It does not automatically cancel the underlying asyncio Task [4]. 2. Manual cancellation: You must explicitly call future.cancel if you wish to stop the execution of the coroutine [1][5]. 3. Cancellation propagation: Calling future.cancel is thread-safe and will schedule a cancellation of the underlying asyncio Task [6][7]. However, because this is an asynchronous operation, the task is not guaranteed to stop immediately [6]. The Task will receive a CancelledError exception during its next cycle in the event loop [3][6]. 4. Known edge cases: There is a known limitation where simply cancelling the future may not guarantee that the underlying task has fully finished its cleanup or reached a terminal state by the time your thread proceeds [4][8]. Because the cancellation is scheduled via the event loop, there is no built-in synchronous way to wait for the task to be fully cleaned up without potential race conditions or leaving the task running temporarily [4]. In summary, future.result(timeout=...) only times out the blocking call in your external thread; you must invoke future.cancel to attempt to stop the coroutine [1][9]. Be aware that the task may continue to run briefly while the cancellation propagates [6].
Citations:
- 1: https://docs.python.org/3/library/asyncio-task.html
- 2: https://docs.python.org/3/library/asyncio-dev.html
- 3: https://docs.python.org/3.10/library/asyncio-task.html
- 4: asyncio.run_coroutine_threadsafe leaves underlying cancelled asyncio task running python/cpython#105836
- 5: https://docs.python.org/release/3.12.1/library/asyncio-task.html
- 6: https://github.com/python/asyncio/blob/master/asyncio/tasks.py
- 7: https://github.com/python/cpython/blob/b35c3791/Lib/asyncio/futures.py
- 8: gh-105836: Fix asyncio.run_coroutine_threadsafe leaving underlying cancelled asyncio task running python/cpython#141696
- 9: https://docs.python.org/release/3.11.0/library/asyncio-task.html
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== events/streams.py relevant lines =="
wc -l events/streams.py
sed -n '1,140p' events/streams.py | cat -n
echo "== dependency version constraints =="
sed -n '1,90p' pyproject.toml | cat -n
if [ -f poetry.lock ]; then
rg -n 'name = "nats"|nats\.py|nats' poetry.lock pyproject.toml -S | head -80
fi
echo "== read-only behavioral model of timeout semantics from source comments =="
python3 - <<'PY'
import asyncio, concurrent, textwrap
print("concurrent.futures timeout docs excerpt:")
print(" TimeoutError from Future.result(timeout=...) indicates wait timed out; underlying Future/Task is not cancelled automatically.")
PYRepository: Jawafdehi/JawafdehiAPI
Length of output: 9448
🌐 Web query:
nats-python v2 connect_timeout raises TimeoutError within client source
💡 Result:
In the NATS Python client (nats-py), the connect_timeout parameter is used to limit the time the client spends attempting to establish the initial connection to a server [1][2]. If the connection cannot be established within this duration, the client generally raises an exception related to connection failure [1][3]. It is important to note that nats.errors.TimeoutError exists within the library's error hierarchy [4][5] and is specifically used in contexts such as request/reply timeouts or JetStream operations [4][6][7]. While the connect_timeout logic is designed to safeguard against network issues, unresponsive servers, or slow handshakes [1], users often find that the client attempts to reconnect based on its internal allow_reconnect and max_reconnect_attempts settings [1][8]. If you wish to force the client to fail strictly after a single attempt when a timeout is reached, you should configure max_reconnect_attempts=1 [1]. Note that there is ongoing evolution in the NATS Python ecosystem; recent migration documentation for core components indicates that some parameter names have been updated (e.g., connect_timeout may be renamed to timeout in newer versions or specific sub-libraries) [9]. You should check your specific version's documentation or source code if you observe different behavior [8][9]. For definitive handling of connection exceptions, you should catch Exception or specific transport errors (like OSError or asyncio.TimeoutError) during the await nats.connect() call [1][3].
Citations:
- 1: Options for
connect()do not work as intended nats-io/nats.py#645 - 2: https://docs.nats.io/using-nats/developer/connecting/connect_timeout
- 3: https://github.com/nats-io/nats.py/blob/main/nats/src/nats/aio/client.py
- 4: https://nats-io.github.io/nats.py/_modules/nats/errors.html
- 5: https://tessl.io/registry/tessl/pypi-nats-py/2.11.0
- 6: [JetStream] Publishing to a subject without stream will fail in nats.errors.TimeoutError nats-io/nats.py#533
- 7: NATS Python example equivalent to 'nats reply' and 'nats request` nats-io/nats.py#558
- 8: nc.connect() API nats-io/nats.py#600
- 9: https://github.com/nats-io/nats.py/blob/e537164138a73f84cf67b4bfa624efa4092ee374/nats-core/MIGRATION.md
🌐 Web query:
nats-py 2.9.3 aio/client.py connect_timeout _process_op
💡 Result:
In the nats-py library (the legacy asyncio NATS client), connect_timeout and _process_op are related to connection management and error handling within nats/aio/client.py [1][2]. connect_timeout: This is a configuration option used during the connection process [1][3]. It defines the maximum duration (in seconds) the client will wait for the transport layer (TCP connection establishment, TLS handshake, and initial NATS protocol handshake) to complete successfully [4][5]. It is not an absolute timeout for the entire connection lifecycle, but specifically for establishing the initial connection to a NATS server [4]. If the connection cannot be established within this time, the attempt is aborted [4]. _process_op_err: While your query referenced _process_op, the relevant method in nats/aio/client.py is typically _process_op_err [1][6]. This internal method is responsible for processing errors that occur while reading or parsing the NATS protocol [1][6]. When a transport error, protocol error, or unexpected end-of-file (EOF) occurs, the client's read loop calls _process_op_err [6]. This method manages the cleanup and, if allowed by the configuration (e.g., allow_reconnect=True), triggers the reconnection logic to attempt a connection to another server in the pool [1][6]. Key points: - The connect_timeout parameter is configurable during the nats.connect call [5]. - _process_op_err is an internal error-handling mechanism invoked during read loop failures [1][6]. - If you encounter issues with connections hanging, verify that your connect_timeout is set appropriately for your network conditions and that you have registered an error_cb to handle and log connectivity issues [4][7].
Citations:
- 1: https://github.com/nats-io/nats.py/blob/main/nats/src/nats/aio/client.py
- 2: https://github.com/nats-io/nats.py/blob/main/CLAUDE.md
- 3: https://github.com/nats-io/nats.py/blob/master/nats/aio/client.py
- 4: Options for
connect()do not work as intended nats-io/nats.py#645 - 5: https://nats-io.github.io/nats.py/modules.html
- 6: Untreated error callbacks when using websockets nats-io/nats.py#361
- 7: https://de.python-3.com/?p=180909
Separate the NATS connection timeout from the startup+stream assertion timeout.
_connect() spends the first 5 seconds on nats.connect(), then must also wait on three sequential js.add_stream() calls, but the wrapping future.result(timeout=STARTUP_TIMEOUT_SECONDS) can time out after connection succeeds. A timeout during stream assertion closes the loop and logs events.connect_failed, leaving the still-running _connect() connection alive because the future timeout does not cancel the task. Use headroom for ensure_streams() / an explicit cancellation path that closes the connection if the coroutine later completes.
🤖 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 `@events/bus.py` around lines 108 - 143, Separate the NATS connection timeout
from the overall startup timeout in _start_locked and _connect: allow
ensure_streams to complete after nats.connect without the outer future expiring
prematurely. If startup still times out or fails, cancel the submitted _connect
coroutine and ensure any connection it creates is closed when cancellation or
late completion occurs, preventing a detached NATS connection from remaining
alive.
| async def ensure_streams(js) -> list[str]: | ||
| """Idempotently assert every stream in :data:`STREAMS`. | ||
|
|
||
| Args: | ||
| js: A JetStream context (``nats.aio.client.Client.jetstream()``). | ||
|
|
||
| Returns: | ||
| The names asserted, in order. | ||
|
|
||
| Raises: | ||
| Whatever the client raises. Unlike publishing, this is NOT best-effort: | ||
| a consumer that cannot see its stream has nothing to do, and should fail | ||
| loudly at startup rather than idle while looking healthy. | ||
| """ | ||
| from nats.js.api import RetentionPolicy, StorageType, StreamConfig | ||
|
|
||
| asserted = [] | ||
| for spec in STREAMS: | ||
| await js.add_stream( | ||
| StreamConfig( | ||
| name=spec.name, | ||
| subjects=list(spec.subjects), | ||
| description=spec.description, | ||
| retention=RetentionPolicy.LIMITS, | ||
| storage=StorageType.FILE, | ||
| max_age=spec.max_age_seconds, | ||
| num_replicas=spec.replicas, | ||
| ) | ||
| ) | ||
| asserted.append(spec.name) | ||
| logger.info( | ||
| "events.stream_asserted", | ||
| stream=spec.name, | ||
| subjects=list(spec.subjects), | ||
| replicas=spec.replicas, | ||
| ) | ||
| return asserted |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== files =="
git ls-files | sed -n '1,200p'
echo "== locate streams.py =="
fd -a 'streams\.py$' . || true
echo "== events/streams.py outline/lines =="
if [ -f events/streams.py ]; then
wc -l events/streams.py
sed -n '1,140p' events/streams.py | cat -n
fi
echo "== nats imports/usages =="
rg -n "nats|jetstream|add_stream|update_stream|StreamSpec|STREAMS|add_stream\(" -S .Repository: Jawafdehi/JawafdehiAPI
Length of output: 15058
🌐 Web query:
NATS JetStream AddStream existing stream same config no-op documentation
💡 Result:
Yes, the AddStream (or CreateStream) operation in NATS JetStream is idempotent [1][2][3]. When you call AddStream with a specific configuration: 1. If the stream does not exist, it will be created [1]. 2. If a stream with the same name already exists, the operation succeeds (as a no-op) only if the provided configuration matches the existing stream's configuration exactly [1][4]. 3. If a stream with the same name exists but the provided configuration differs, the operation will fail and return an error (typically nats.ErrStreamNameAlreadyInUse) [4][5]. Because of this behavior, you do not need to manually check for the existence of a stream before calling AddStream, provided that the configuration you supply is the desired final state [4]. If you intend to ensure a stream exists with a specific configuration regardless of its previous state (or to update it if it exists), you would typically need to perform an existence check or use an "upsert" pattern (e.g., attempt an update, and if it fails with a "not found" error, perform a create) [4][6].
Citations:
- 1: https://docs.nats.io/using-nats/developer/develop_jetstream/streams.md
- 2: Should programmatically adding a stream be idempotent? nats-io/nats-server#2527
- 3: https://docs.nats.io/using-nats/developer/develop_jetstream
- 4: The proper way to check if the stream exists nats-io/nats.go#1087
- 5: Add detail to JSStreamNameExistErr nats-io/nats-server#3273
- 6: Add create_or_update_stream to JetStream nats-io/nats.py#964
🌐 Web query:
nats.io nats-client-go JSStreamAlreadyExists update_stream StreamConfig subject changes
💡 Result:
In the nats-client-go library, you can update a stream's configuration, including its subjects, using the UpdateStream method [1][2][3]. If you are encountering an error related to "already exists" (e.g., JSStreamAlreadyExists), it often occurs when attempting to create a stream that already exists rather than updating it [4][3]. When modifying an existing stream, you must use the UpdateStream operation, which targets the stream by name and applies the new StreamConfig [1][4]. Key points for updating streams in nats-client-go: 1. Use UpdateStream: The UpdateStream function is explicitly designed for this purpose [1][5]. It sends a request to the JetStream API endpoint $JS.API.STREAM.UPDATE.<stream_name> [6][7][8]. 2. Provide Full Configuration: When calling UpdateStream, you must provide a StreamConfig object that contains the desired new state [1][2]. Note that you typically need to include all fields required for the configuration, as the update generally replaces or reconciles the existing configuration on the server side [2][5]. 3. Handling Errors: If you receive errors such as ErrStreamNotFound, ensure the stream name provided in the configuration matches an existing stream exactly [1][9]. Errors like JSStreamAlreadyExists (or similar "already in use" errors) should not occur when using the correct UpdateStream path; if they do, verify that you are not accidentally calling CreateStream instead of UpdateStream [9][4][3]. 4. Subject Changes: You can change the subjects by providing the updated list in the Subjects field of the StreamConfig passed to UpdateStream [2][5]. If you are using the modern JetStream API (the jetstream package in nats.go), the pattern is: // Assuming 'js' is your JetStream context and 's' is your stream handle or via the context err:= js.UpdateStream(ctx, jetstream.StreamConfig{ Name: "YOUR_STREAM_NAME", Subjects: []string{"NEW.SUBJECT.*"}, //... include other necessary config fields }) [4][3]
Citations:
- 1: https://github.com/nats-io/nats.go/blob/master/jsm.go
- 2: https://github.com/nats-io/nats.go/blob/5ec9af538e4a22478326c81b329422ac35d52a3e/jsm.go
- 3: https://github.com/nats-io/nats.go/tree/main/jetstream
- 4: https://github.com/nats-io/nats.go/blob/51412b787fa04e0e3c469145d7d61dfef79cf991/jetstream/README.md
- 5: https://github.com/nats-io/nats.go/blob/4b75fc59ae30774622e53617b91f9a03215fcccd/jsm.go
- 6: https://github.com/nats-io/nats.docs/blob/master/using-nats/jetstream/nats_api_reference.md
- 7: https://docs.nats.io/reference/reference-protocols/nats_api_reference
- 8: https://github.com/nats-io/nats.go/blob/3c9d1b174a725d3d174c4081af8fcb778e313387/js.go
- 9: https://github.com/nats-io/nats.go/blob/main/jserrors.go
🌐 Web query:
nats-py 2.15.0 JavaScriptContext add_stream update_stream source
💡 Result:
In nats-py 2.15.0, stream management operations like adding or updating a stream are handled by the JetStreamManager, not the standard JetStreamContext used for publishing and subscribing [1][2][3]. To add or update a stream, you must first obtain an instance of the JetStreamManager from your NATS connection [3]. Management Methods: 1. add_stream(config=None, params): Creates a new stream. This is an idempotent operation [1][2][4]. 2. update_stream(config=None, params): Updates the configuration of an existing stream [1][2]. Example Usage: import asyncio import nats from nats.js.api import StreamConfig async def main: nc = await nats.connect("nats://localhost:4222") # Obtain the JetStream Manager jsm = await nc.jetstream_manager # Example: Add a stream # You can pass a StreamConfig object or keyword arguments await jsm.add_stream(name="my-stream", subjects=["my-subject.*"]) # Example: Update a stream # Note: Only specific fields in StreamConfig are mutable via update_stream await jsm.update_stream(name="my-stream", max_msgs=1000) await nc.close asyncio.run(main) The JetStreamContext (obtained via nc.jetstream) is primarily used for operational tasks like publishing messages, creating consumers, and fetching messages, whereas the JetStreamManager is explicitly designed for administrative actions on streams and consumers [5][2][3].
Citations:
- 1: https://nats-io.github.io/nats.py/_modules/nats/js/manager.html
- 2: https://nats-io.github.io/nats.py/modules.html
- 3: https://github.com/nats-io/nats.docs/blob/master/using-nats/developing-with-nats/js/context.md
- 4: https://docs.nats.io/using-nats/developer/develop_jetstream/streams
- 5: https://pypi.org/project/nats-py/2.15.0/
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== event subscription files =="
git ls-files events| sed -n '1,200p'
echo "== test_bus relevant lines =="
sed -n '1,190p' events/tests/test_bus.py | cat -n
echo "== bus relevant lines =="
sed -n '11,220p' events/bus.py | cat -nRepository: Jawafdehi/JawafdehiAPI
Length of output: 17924
🌐 Web query:
nats-py 2.15.0 nc.jetstream returns JetStreamManager or context add_stream available
💡 Result:
In nats-py 2.15.0, the JetStreamManager class does indeed provide an add_stream method [1]. This method is used to create a stream by sending a request to the NATS JetStream management API [1]. You can access the JetStreamManager through your NATS connection object [2]. While the typical JetStream context (returned by nc.jetstream) is primarily used for publishing and subscribing to messages, the JetStreamManager is explicitly designed for administrative tasks such as stream and consumer management [2]. Example usage of add_stream via JetStreamManager: import asyncio import nats from nats.js.api import StreamConfig async def main: nc = await nats.connect("localhost") # Access the JetStream management API jsm = await nc.jetstream_manager # Create a stream using the manager stream_info = await jsm.add_stream( config=StreamConfig(name="example-stream", subjects=["example-subject"])) print(f"Stream created: {stream_info.config.name}") await nc.close asyncio.run(main) Note that while the JetStream context (js) in some older documentation examples historically showed a convenience method js.add_stream [3][4], the recommended and more robust way to handle stream configuration and management in current versions is via the JetStreamManager [1][2]. Always ensure your stream configuration adheres to the NATS naming and validation requirements (e.g., no whitespace or special characters like. or *) which the manager now explicitly validates before sending the request [1][5].
Citations:
- 1: https://nats-io.github.io/nats.py/_modules/nats/js/manager.html
- 2: https://nats-io.github.io/nats.py/modules.html
- 3: https://pypi.org/project/nats-py/2.15.0/
- 4: https://github.com/nats-io/nats.py/blob/main/README.md
- 5: https://github.com/nats-io/nats.py/releases/tag/v2.15.0
🌐 Web query:
nats.py JetStreamContext add_stream update_stream no-op existing stream
💡 Result:
In nats.py, the add_stream method in JetStreamContext is designed to be idempotent, meaning it will succeed if the stream already exists, provided the existing configuration matches the requested configuration [1][2][3][4]. If a stream with the same name exists but has a different configuration, the call will typically fail, as it cannot implicitly update the stream's settings [1][2][4]. For scenarios where you want to ensure a stream is created or updated to a desired configuration regardless of its current state, nats.py now includes a create_or_update_stream method [5]. This method functions as an idempotent upsert: it attempts to update the stream first and falls back to creating it if it does not already exist [5]. Summary of Key Methods: - add_stream: Idempotent create. Fails if the stream exists but has a conflicting configuration [1][2]. - update_stream: Explicitly updates an existing stream's configuration [6]. - create_or_update_stream: Idempotent upsert. Automatically handles both create and update operations [5]. If you encounter unexpected errors when using these methods, ensure your stream name follows the required format. Stream names cannot contain whitespace, '.', '*', '>', path separators, or non-printable characters [6]. Providing an invalid name can lead to library-level errors or timeouts [7][8].
Citations:
- 1: https://docs.nats.io/using-nats/developer/develop_jetstream/streams.md
- 2: https://docs.nats.io/using-nats/developer/develop_jetstream/streams
- 3: https://docs.nats.io/using-nats/developer/develop_jetstream
- 4: Should programmatically adding a stream be idempotent? nats-io/nats-server#2527
- 5: Add create_or_update_stream to JetStream nats-io/nats.py#964
- 6: https://nats-io.github.io/nats.py/_modules/nats/js/manager.html
- 7: with Jetstream, get "nats: timeout" when Add Stream or pull_subscribe to an existing stream nats-io/nats.py#437
- 8: Passing wrong stream name to
add_streamthrowserrors.TimeoutErrornats-io/nats.py#471
🌐 Web query:
site:nats-io.github.io/ JetStreamManager create_or_update_stream update_stream add_stream nats-py
💡 Result:
In the nats-py library, the JetStreamManager class provides the administrative interface for managing JetStream resources [1][2]. There is no built-in method named create_or_update_stream in the JetStreamManager API [3][1][2]. Instead, management is handled through distinct methods for adding and updating streams: 1. add_stream(config: Optional[api.StreamConfig] = None, params) -> api.StreamInfo: This method is used to create a new stream [3][1]. It accepts an optional StreamConfig object and additional keyword arguments that are passed to the configuration object via its evolve method [3]. It validates the stream name (e.g., ensuring no invalid characters like whitespace or wildcards) before sending the creation request to the NATS server [3]. 2. update_stream(config: Optional[api.StreamConfig] = None, params) -> api.StreamInfo: This method is used to modify an existing stream [3][1]. Similar to add_stream, it takes an optional StreamConfig and keyword arguments to update the stream's configuration [3]. To implement a "create or update" pattern in your application code, you would typically attempt to retrieve the stream information first (using stream_info), check if it exists, and then call either add_stream or update_stream accordingly [3][2]. Top results: [3][1][2]
Citations:
- 1: https://nats-io.github.io/nats.py/modules.html
- 2: https://nats-io.github.io/nats.py/index.html
- 3: https://nats-io.github.io/nats.py/_modules/nats/js/manager.html
Use update_stream for config drift, and only raise on true broker/setup failures.
NATS accepts add_stream as a no-op only when the existing stream’s config matches exactly; otherwise it fails and leaves ensure_streams() fatal at startup. That means a future StreamSpec change for subjects, lifecycle limits, or replicas will wedge every existing broker deployment. Try add_stream() first, but fallback to js.update_stream(...) when the stream already exists with a conflicting config before propagating other errors.
🤖 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 `@events/streams.py` around lines 65 - 101, The ensure_streams function should
tolerate expected configuration drift: attempt add_stream first, and when it
fails because the stream already exists with conflicting configuration, fetch or
reuse the existing stream definition and call js.update_stream with the current
StreamConfig. Preserve propagation of genuine broker or setup failures, and keep
assertion logging and return behavior unchanged.
… dependency The e2e job caught what the unit suite structurally could not: `events` is already a top-level import name, shipped by the `Events` distribution that `opensearch-py` pulls in transitively. From a source checkout this looks fine — the repo root precedes site-packages on sys.path, so our package shadowed the dependency and all 2603 tests passed. The Docker image installs the app as a WHEEL INTO site-packages, where both land in the same directory and collide, so `manage.py migrate` exited 1 before the stack could come up. Renamed to `case_events`, matching the sibling `case_proposals`. A two-word name is also far less likely to be claimed by some future transitive dependency than a bare noun. Added tests/test_app_package_names.py so this cannot recur: it asserts no first-party app shares a top-level name with an installed distribution, which is the same question the wheel install asks. "First party" is defined by the package directory existing in the repo, not by a name prefix — otherwise legitimately-named third-party apps (auditlog from django-auditlog, corsheaders from django-cors-headers) read as false positives. It also carries a guard-the-guard test: if packages_distributions() ever stops reporting owners, the collision check would pass vacuously forever, so a second test asserts it still sees the known-colliding `events` name. Verified the built wheel now exposes case_events and no bare events/. Co-Authored-By: Claude <noreply@anthropic.com>
There was a problem hiding this comment.
Your trial has ended. Reactivate Greptile to resume code reviews.
…t be missed The real cause of the failing build, and my previous commit message got it wrong. `ModuleNotFoundError: No module named 'case_events'` at collectstatic: the Dockerfile COPYs each app directory by name and the new app was never added, so it simply was not in the image. Correcting the earlier diagnosis. The `events` name collision was real but was NOT the trigger — it was what made the trigger unreadable. With the app absent from the image, `import events` did not raise; it silently resolved to the third-party `Events` package that opensearch-py pulls in transitively (verified: it has no `apps` submodule and no `bus`), so a missing COPY surfaced as a strange ImportError in an unrelated module. The rename still stands on its own merits, but this is the fix. Adding an app means editing THREE lists — INSTALLED_APPS, the wheel `packages`, and the Dockerfile COPY block — and none of them fail at test time, because the unit suite runs from a source checkout where every directory is importable regardless. So the omission is invisible until an image is built. tests/test_app_package_names.py now asserts all three agree, and each check was mutation-verified: deleting the COPY line fails the test with an actionable message naming the app and the exact line to add. Left the Dockerfile list explicit rather than switching to `COPY . .`, which would drag the venv, .git and fixtures into the image. Co-Authored-By: Claude <noreply@anthropic.com>
There was a problem hiding this comment.
Your trial has ended. Reactivate Greptile to resume code reviews.
Two follow-up commits — and a correction to my first diagnosisCI initially failed the The actual cause: a missing What I first blamed, and why it was only half right. The app was originally named I've kept the rename to Why the unit suite couldn't have caught either. Adding an app means editing three lists —
Left the COPY list explicit rather than switching to All four checks green now. |
PromptSpec held its system prompt as an inline string and built its content
block with a Python callable. That keeps prompt wording in Python, where
reviewing a change means reading a diff of an f-string. Prompts are prose and
should be reviewable as prose, so the text moves into
<app>/prompt_templates/*.md and a spec now names two template paths.
Rendering goes through llm/templating.py: a DEDICATED Django template engine,
not the one in settings.TEMPLATES. Two of that engine's defaults are actively
wrong for prompts and both fail silently.
Autoescaping had to go. The hazard is not the prompt wording — Django escapes
interpolated values, not literal template text — it is the data. review/judge.py
passes json.dumps(case_summary, indent=2); escaped, that reaches the model as a
wall of " and stops being JSON. Source excerpts carry quotes, case titles
carry ampersands. The prompt still looks like a prompt and nothing raises.
Undefined variables had to stop rendering as "". A renamed context key would
otherwise ship a prompt with a hole in it, and the failure surfaces much later
as a bad record with no obvious cause. The engine gets a NUL-delimited
string_if_invalid sentinel and render_prompt refuses to return any string still
containing it, naming the variable that failed.
That sentinel does not cover tags: Django resolves {% if missing %} to falsy and
{% for x in missing %} to empty without consulting string_if_invalid. Variables
used only inside a tag must be declared in PromptSpec.required, which is checked
before rendering. A test documents the gap and proves `required` closes it.
Both guards are mutation-verified: flipping autoescape on fails 5 tests,
dropping the sentinel fails 5 others. An earlier draft of the autoescape tests
put the HTML in the template body and passed under the mutation, which is what
established the value-vs-literal distinction now recorded in the docstring —
every test here puts the hostile characters in a context value.
Verified that .md files actually survive `uv build`, since a data file excluded
from the wheel would be exactly the kind of image-only breakage that already hit
this branch once. They do; the Dockerfile COPYs llm/ wholesale.
Convention notes: the directory is prompt_templates/, not prompts/, because
llm/prompts.py exists and a module and package of the same name cannot coexist.
Dirs are discovered from the app registry, so an app owns its prompts.
Scope is the registry only. review/judge.py and the casework/enrich_* prompts
are NOT migrated: they work, and several of their tests assert on the Python
constants directly (test_enrich_timeline.py parses the source AST for them).
Co-Authored-By: Claude <noreply@anthropic.com>
There was a problem hiding this comment.
Your trial has ended. Reactivate Greptile to resume code reviews.
1 participant
Phase 1 of the case-enrichment event bus — the two workstreams that are pure code and merge safely with no broker running. Design docs:
DESIGN.md,PHASE-1-PLAN.md,RUNTIME-TOPOLOGY.md(scratch, not in this repo).No migration.
eventshas no models, andJob.kindis already a free-formCharField.makemigrations --checkconfirms no changes detected. Given migrations here are run manually, that removes the deploy-ordering risk from this PR entirely.Off by default.
NATS_URLis unset, so every publish is a logged no-op and no thread is ever started. Dev and CI need no broker. Setting the variable is also the entire rollback — no image change.events/— the busEnvelope, subject vocabulary, JetStream stream topology, and a sync→async publisher (
nats-pyis asyncio-only; Django's request path is not). Three properties it exists to guarantee:Streams are asserted from code rather than declared in YAML. NATS has no CRD,
add_streamis upsert-like, and this keeps the topology beside the code that depends on it — so a fresh or local environment needs no bootstrap step. Unlike publishing, this is not best-effort: a consumer that can't see its stream should fail loudly at startup rather than idle while looking healthy.case_proposals/publish.py— the first real producerApprove/reject emit
jaw.case.update.{approved,rejected}viatransaction.on_commit, mirroring the existing_schedule_reindex/_schedule_material_visibilityhooks next door. Registered inside the atomic block so a rollback discards the callback, but fired only after a successful commit — so a subscriber can never read the case before the write is visible.The decision's
dedup_keyis keyed on the decision (proposal:<pk>:<status>), deliberately distinct from the fact'sdedup_key, which rides in the payload asfact_dedup_key. Re-publishing a decision collapses in JetStream without suppressing a genuinely different event about the same fact.llm/prompts.py— PromptSpecA named, versioned record of (system prompt, content builder, tier, max tokens) that
invoke_jsonis called with. Version matters because "which prompt produced this wrong intent?" is the first question asked of any LLM-generated record. Deliberately not a templating engine or a DB-backed CMS, and migratingreview/judge.pyonto it is out of scope.It validates
tier, which is not decoration:routing.provider_for_tierresolves"premium"and treats everything else as cheap, so a typo ("premuim","strong") would silently downgrade the model rather than raise. Intent generation must stay on the strong tier — weak models fail to emit patch-shaped JSON.get()raises on an unknown name, deliberately diverging fromjobs.registry.get, which returns a default. A job kind has a sensible default policy; a prompt does not.Testing
51 new tests. The load-bearing ones assert an approval still succeeds and still applies its intent both with no broker configured and with the publisher raising — that property is the whole point of the design.
Verified non-vacuous by mutation: removing the
schedule_decision_event(proposal)call fails exactly the two tests that assert publishing, and none of the others. On-commit paths are exercised withdjango_capture_on_commit_callbacks(execute=True), without which they'd pass vacuously.uv run pytest -q --tb=short --ignore=integration-tests→ 2603 passed, 4 skippeduv run ruff check→ clean (ruff formatis intentionally not gated in this repo, so no reformatting)manage.py check→ clean apart from the pre-existing treebeard/Wagtail warningsNot in this PR
Workstream A (the NATS manifests) touches the cluster and is unapplied. D (consumers) and E (producers) depend on it. Two unrelated fixes surfaced while validating the runtime docs and are deliberately left out:
ENV LANG=C.UTF-8in the base image, andapt-get install antiwordto soften the amd64 pin — seeRUNTIME-TOPOLOGY.md§4.🤖 Generated with Claude Code
Summary by CodeRabbit
New Features
Reliability
Tests