From 4fd8479318388a0d9fa8252945f45ea78f151ef1 Mon Sep 17 00:00:00 2001 From: Krish Garg Date: Sun, 12 Jul 2026 03:22:31 -0700 Subject: [PATCH 1/3] fix(endpoints): unique stable per-pipe endpoint URLs Advertised chat/dropper/webhook URLs now embed the task identity as {host}/{type}/{project_id}/{source}; additive path routes accept data POSTs and resolve the task from the path. Legacy routes and ?auth= resolution unchanged. Fixes rocketride-ai/rocketride-saas#262 Co-Authored-By: Claude Fable 5 --- nodes/src/nodes/webhook/IEndpoint.py | 10 +++++----- nodes/src/nodes/webhook/README.md | 2 +- packages/ai/src/ai/modules/task/task_engine.py | 8 ++++++++ .../ai/src/ai/modules/task_http/__init__.py | 2 ++ .../ai/src/ai/modules/task_http/task_data.py | 10 ++++++++++ .../content-static/examples/webhook-pipeline.md | 17 +++++++++++------ 6 files changed, 37 insertions(+), 12 deletions(-) diff --git a/nodes/src/nodes/webhook/IEndpoint.py b/nodes/src/nodes/webhook/IEndpoint.py index 1ec7e7fa8..12c35df27 100644 --- a/nodes/src/nodes/webhook/IEndpoint.py +++ b/nodes/src/nodes/webhook/IEndpoint.py @@ -64,9 +64,9 @@ async def _startup(self): # These should NOT be replacable strings!!! info = { 'button-text': 'Chat now', - 'button-link': '{host}/chat?auth={public_auth}', + 'button-link': '{host}/chat/{project_id}/{source}?auth={public_auth}', 'url-text': 'Chat interface URL', - 'url-link': '{host}/chat', + 'url-link': '{host}/chat/{project_id}/{source}', 'auth-text': 'Public Authorization Key', 'auth-key': '{public_auth}', 'token-text': 'Private Token', @@ -88,9 +88,9 @@ async def _startup(self): # These should NOT be replacable strings!!! info = { 'button-text': 'Drop now', - 'button-link': '{host}/dropper?auth={public_auth}', + 'button-link': '{host}/dropper/{project_id}/{source}?auth={public_auth}', 'url-text': 'Dropper interface URL', - 'url-link': '{host}/dropper', + 'url-link': '{host}/dropper/{project_id}/{source}', 'auth-text': 'Public Authorization Key', 'auth-key': '{public_auth}', 'token-text': 'Private Token', @@ -115,7 +115,7 @@ async def _startup(self): } info = { 'url-text': url_text_map[self.endpoint.logicalType], - 'url-link': '{host}/webhook', + 'url-link': '{host}/webhook/{project_id}/{source}', 'auth-text': 'Public Authorization Key', 'auth-key': '{public_auth}', 'token-text': 'Private Token', diff --git a/nodes/src/nodes/webhook/README.md b/nodes/src/nodes/webhook/README.md index 0db293bab..b9a0dde5c 100644 --- a/nodes/src/nodes/webhook/README.md +++ b/nodes/src/nodes/webhook/README.md @@ -12,7 +12,7 @@ Stands up its own HTTP endpoint and forwards incoming data into the attached pip The server is a **FastAPI / Uvicorn** wrapper (`ai.web.WebServer`). The engine passes the bind address on the command line (`--data_host`, `--data_port`, defaulting to `localhost:5567`); the node loads the `data` module, which registers a `/task/data` websocket as the data plane between the public endpoint and the pipeline. The node keeps running until the pipeline is stopped, the source task completes only when the server exits. -After the pipeline starts, the Project Log displays the interface URL, the public authorization key, and the private token, so callers know how to reach the endpoint (the chat link form is `{host}/chat?auth={public_auth}`). +After the pipeline starts, the Project Log displays a stable, pipe-specific interface URL, the public authorization key, and the private token, so callers know how to reach the endpoint. The URL forms are `{host}/chat/{project_id}/{source}?auth={public_auth}`, `{host}/dropper/{project_id}/{source}?auth={public_auth}`, and `{host}/webhook/{project_id}/{source}`. The legacy `/chat`, `/dropper`, `/webhook`, and `/task/data` URLs continue to work for existing integrations. --- diff --git a/packages/ai/src/ai/modules/task/task_engine.py b/packages/ai/src/ai/modules/task/task_engine.py index 94a764965..de9084398 100644 --- a/packages/ai/src/ai/modules/task/task_engine.py +++ b/packages/ai/src/ai/modules/task/task_engine.py @@ -979,6 +979,10 @@ def _update_status(self, message: Dict[str, Any]) -> None: # Handle string notes - simple replacement note = note.replace('{token}', self.token) note = note.replace('{public_auth}', self.public_auth) + if self.project_id is not None: + note = note.replace('{project_id}', str(self.project_id)) + if self.source is not None: + note = note.replace('{source}', str(self.source)) self._status.notes.append(note) elif isinstance(note, dict): # Handle dict notes - walk through and replace in all string values @@ -988,6 +992,10 @@ def _update_status(self, message: Dict[str, Any]) -> None: # Replace tokens in string values value = value.replace('{token}', self.token) value = value.replace('{public_auth}', self.public_auth) + if self.project_id is not None: + value = value.replace('{project_id}', str(self.project_id)) + if self.source is not None: + value = value.replace('{source}', str(self.source)) processed_note[key] = value self._status.notes.append(processed_note) else: diff --git a/packages/ai/src/ai/modules/task_http/__init__.py b/packages/ai/src/ai/modules/task_http/__init__.py index 59a189793..e94a42ab0 100644 --- a/packages/ai/src/ai/modules/task_http/__init__.py +++ b/packages/ai/src/ai/modules/task_http/__init__.py @@ -23,11 +23,13 @@ def initModule(server: WebServer, config: Dict[str, Any]): # Preferred routes (POST) server.add_route('/task', task_Execute, methods=['POST']) server.add_route('/task/data', task_Data, methods=['POST']) + server.add_route('/task/data/{project_id}/{source}', task_Data, methods=['POST']) server.add_route('/task', task_Status, methods=['GET']) server.add_route('/task', task_Cancel, methods=['DELETE']) # Alias to /task/data server.add_route('/webhook', task_Data, methods=['POST']) + server.add_route('/webhook/{project_id}/{source}', task_Data, methods=['POST']) # Deprecated routes (PUT) - kept for backward compatibility server.add_route('/task', task_Execute, methods=['PUT'], deprecated=True) diff --git a/packages/ai/src/ai/modules/task_http/task_data.py b/packages/ai/src/ai/modules/task_http/task_data.py index 54094ed88..69533a41b 100644 --- a/packages/ai/src/ai/modules/task_http/task_data.py +++ b/packages/ai/src/ai/modules/task_http/task_data.py @@ -469,6 +469,16 @@ async def task_Data( client = None try: + if token is None: + project_id = request.path_params.get('project_id') + source = request.path_params.get('source') + if project_id is not None and source is not None: + try: + control = request.app.state.task.get_task_control_by_project(project_id, source) + token = control.token + except Exception: + pass + # Get the WebServer instance from application state server: WebServer = request.app.state.server port = server.get_port() diff --git a/packages/docs/content-static/examples/webhook-pipeline.md b/packages/docs/content-static/examples/webhook-pipeline.md index 1f1afd4ea..496ecc8d6 100644 --- a/packages/docs/content-static/examples/webhook-pipeline.md +++ b/packages/docs/content-static/examples/webhook-pipeline.md @@ -15,6 +15,7 @@ Save this as `chat.pipe`: ```json { + "project_id": "webhook-example", "nodes": [ { "id": "source_1", @@ -57,7 +58,7 @@ Output: ```text Webhook ready - system is ready to accept requests - URL: http://localhost:5567/task/data + URL: http://localhost:5567/webhook/webhook-example/source_1 Auth: Token: ``` @@ -67,7 +68,7 @@ Keep the terminal open — the pipeline runs until you stop it. ## Send a question ```bash -curl -X POST http://localhost:5567/task/data \ +curl -X POST "http://localhost:5567/webhook/webhook-example/source_1" \ -H "Authorization: Bearer " \ -H "Content-Type: text/plain" \ -d "What is the capital of France?" @@ -92,13 +93,17 @@ Restart the pipeline. The engine prints a chat URL: ```text Chat ready - system is ready to accept questions - URL: http://localhost:5567/chat?auth= + URL: http://localhost:5567/chat/webhook-example/source_1?auth= ``` Open that URL in a browser and start typing. The `?auth=` query parameter is a -convenience for the browser; the endpoint also accepts the key in an -`Authorization: Bearer ` header, the same scheme the webhook -endpoint uses. +convenience for supplying the credential to the browser; the Chat UI stores it +in session storage for subsequent page loads. Webhook requests use the same key +in an `Authorization: Bearer ` header. + +Each displayed URL includes the project and source identifiers, so it remains +specific to this pipeline. The legacy `/webhook`, `/task/data`, `/chat`, and +`/dropper` URLs continue to work for existing integrations. ## Add a system prompt From 7663b8477cd8e1c09b4ea4ace881c007f619db85 Mon Sep 17 00:00:00 2001 From: Krish Garg Date: Sun, 12 Jul 2026 14:44:23 -0700 Subject: [PATCH 2/3] fix(task-http): only treat missing tasks as token-resolution fallback Co-Authored-By: Claude Fable 5 --- packages/ai/src/ai/modules/task_http/task_data.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/packages/ai/src/ai/modules/task_http/task_data.py b/packages/ai/src/ai/modules/task_http/task_data.py index 69533a41b..47f5175e8 100644 --- a/packages/ai/src/ai/modules/task_http/task_data.py +++ b/packages/ai/src/ai/modules/task_http/task_data.py @@ -476,7 +476,10 @@ async def task_Data( try: control = request.app.state.task.get_task_control_by_project(project_id, source) token = control.token - except Exception: + except RuntimeError: + # Documented missing-task case ('Your pipeline is not + # running') — fall back to token=None so the request takes + # the normal auth/404 path. Anything else propagates. pass # Get the WebServer instance from application state From f0d042f35783aafd697d1fba92e7f64719128eb3 Mon Sep 17 00:00:00 2001 From: Krish Garg Date: Sun, 12 Jul 2026 20:13:04 -0700 Subject: [PATCH 3/3] fix(task): guard identity placeholders for partially-initialized tasks Co-Authored-By: Claude Fable 5 --- .../ai/src/ai/modules/task/task_engine.py | 22 ++++++++++++------- 1 file changed, 14 insertions(+), 8 deletions(-) diff --git a/packages/ai/src/ai/modules/task/task_engine.py b/packages/ai/src/ai/modules/task/task_engine.py index de9084398..582396239 100644 --- a/packages/ai/src/ai/modules/task/task_engine.py +++ b/packages/ai/src/ai/modules/task/task_engine.py @@ -979,10 +979,14 @@ def _update_status(self, message: Dict[str, Any]) -> None: # Handle string notes - simple replacement note = note.replace('{token}', self.token) note = note.replace('{public_auth}', self.public_auth) - if self.project_id is not None: - note = note.replace('{project_id}', str(self.project_id)) - if self.source is not None: - note = note.replace('{source}', str(self.source)) + # getattr: partially-initialized tasks (and test stubs) + # may not carry the identity attributes at all. + project_id = getattr(self, 'project_id', None) + source = getattr(self, 'source', None) + if project_id is not None: + note = note.replace('{project_id}', str(project_id)) + if source is not None: + note = note.replace('{source}', str(source)) self._status.notes.append(note) elif isinstance(note, dict): # Handle dict notes - walk through and replace in all string values @@ -992,10 +996,12 @@ def _update_status(self, message: Dict[str, Any]) -> None: # Replace tokens in string values value = value.replace('{token}', self.token) value = value.replace('{public_auth}', self.public_auth) - if self.project_id is not None: - value = value.replace('{project_id}', str(self.project_id)) - if self.source is not None: - value = value.replace('{source}', str(self.source)) + project_id = getattr(self, 'project_id', None) + source = getattr(self, 'source', None) + if project_id is not None: + value = value.replace('{project_id}', str(project_id)) + if source is not None: + value = value.replace('{source}', str(source)) processed_note[key] = value self._status.notes.append(processed_note) else: