Future-proof your skills and escape the tech hamster wheel with the Linux, Python, vim & git stack (LPvg) including NixOS, Jupyter, FastHTML / HTMX and an AI stack to resist obsolescence. Follow along as I debunk peak data theory and develop Pipulate, the next generation free AI SEO tool.

Grok3 is Free Until Our Servers Melt

This article details my journey in creating a new workflow, Hello Webpage, by modifying a template using the WET software methodology, where I'm intentionally allowing code duplication for greater flexibility. I was genuinely impressed when xAI's Grok3 provided the exact working code I needed on its first attempt and then tried helping me identify and fix a tricky bug related to Pipeline ID collisions across different workflow applications, but came up short. Despite some moments where Grok3 seemed to freeze up during its responses, it ultimately persevered, highlighting its unique resilience compared to other AI models I've used. This experience has reinforced the potential of this new AI as a valuable partner in my coding and creative endeavors.

Grok3’s Unique Approach to Conversation History

Long conversation histories aren’t easy for Grok3. It’s not precisely the 128K token window that was so well advertised by Google for Gemini 1.5, and which is now common even for the locally running Gemma3 under Ollama. No, with Grok3 it’s something different. It gives a good college try until it’s crushed under the length of the discussion! Is that a problem with Grok? I think not. You just have to live with the output getting slower and slower.

The Evolution of Grok3’s Interface

When I first started interacting with Grok3 on X.com, I was furious because the copy-text feature of responses wasn’t formatting markdown, which is a complete showstopper in my document-everything (and publish half of it 😆) workflow. But now they have the grok.com domain and are advertising it on what looks like a model dropdown-selector, and on that site markdown is formatted properly! And from that moment, I started using Grok3 at least as the 2nd rubber duck.

Grok3 as a Unique Rubber Duck Debugging Partner

I started a very fruitful discussion with Grok3 at the end of yesterday’s article. Grok3’s attitude, status as an outlier and willingness to go where other models won’t is an asset and a boon to any project. If for nothing else, as a different rubber ducky to try when rubber ducking – a process of bouncing ideas off themselves (aka talking to the rubber duck). When using AIs in this capacity, these subtle model differences are import, especially when living in the world of anti-patterns like FastHTML, HTMX, localhost and Nix.

The Cost of Being Different: xAI’s Infrastructure Investment

The fact that Grok3’s output is different and potentially more useful because of its bad-boy attitude on things different than anti-woke-virus conversation, things like programming, is such a nuanced point. And that it’s being given away so that the point can sink-in with the X.com-using public is… well remarkable yes, but remarkably expensive! What amounts to the largest supercomputer in the world was built for xAI in Memphis, Tennessee. They’re reportedly to ~200K GPUs but are aiming for a million. 1 million H100 GPUs plus infrastructure could be $60 billion or more.

That’s not even counting NVIDIA’s Blackwell Rubin and Feynman architectures (including the B200 and B300 series). It’s also not counting all the money poured into continuing developing the line of Tesla AI4 custom chips in their own Dojo supercomputer architecture. Whether xAI goes with NVIDIA chips or their own custom stuff, it comes up massive bucks.

The Economics of Free AI Access

At bare minimum, xAI and Grok3 is like buying Twitter twice!

Free. Or at least the first dose is free. And that first does has to be that differentiated from Gemini, Claude, ChatGPT, Perplexity and the rest of the pack, so much so that they get your $20/mo. subscription, which seems to be the sweet spot that the general consumer-level public is willing to spend. And with only about 300-million people in the United States, let’s say only 1-in-3 is a qualified prospect, that’s 100-million people. The calculation is:

100,000,000 potential customers x $20/month x 12 months = $24,000,000,000 (24 billion dollars per year)

24 billion/year if every single qualified customer pays their $20/mo AI tax. Is that a lot? I don’t know. But Elon’s statement about giving it away for free or until their servers melt down gives you some sense of the determination to get their piece of whatever pie is up for grabs. And that’s just the financial part – not even mentioning the hearts-and-minds issue. Whatever you pick as your corporate AI-buddy is going to shape your mind at least as much as one of the 5 people… hmmm, that concept:

“You are the average of the five people you spend the most time with,” and it’s attributed to motivational speaker Jim Rohn. The people we spend the most time with greatly influence our thoughts, beliefs, habits, behaviors, and life choices.

…and your chosen corporate AI of choice is at least one of them! Forget about the YouTube algorithm. These things are the teachers and mentors directly talking to our children once they gain enough agency to find them and ask.

The Importance of Markdown Formatting in AI Interfaces

Thanks, Google. I turned on my Google search history to get AI Mode that’s part of Google Labs to get the 5-people quote you see above. The above answer came from there, but you know what? No markdown copy-text link! Ugh! I just thumbs-downed the response and sent the same feedback to Google as I did to xAI: if you don’t have a copy-text button for the response that gives you formatted markdown, it might as well not exist. I told it to give them a screenshot too, to differentiate my feedback.

Comparing AI Models’ Response Patterns

Anyway, that was a fun digression. But important. Google’s Gemini 2.0 Advanced still won’t acknowledge who the US President is without asking it for the latest news headlines sufficiently so that a preponderance of evidence would put it in almost the AI analogue to nearly explosive cognitive dissonance, haha! It will eventually acknowledge it, but then it clams up as if going into some defensive mode. The nuances of each AI is so interesting! Anyway, back to Grok3, and its coding advice in my next-step.

Testing LLMs with Anti-Pattern Code Systems

This is actually still a continuous conversation with Grok3 carried over from yesterday, with the context-window growing bigger and bigger with full code-bases dumped on it. And now, this is a request for the first modification of the Hello World template example Pipulte AI workflow, which is what you call a WET code system (as opposed to DRY), and it’s a really great anti-pattern to test the LLMs with. The reason for WETness is to allow extreme customization without being inhibited by the tangled web of constraints the alternative, object oriented programming and inheritance from template-like superclasses puts on you.

So, from where we left off…

Me: That’s a really smart next step, Grok! But instead of Hello World, I think it should be Hello Webpage! And rather than Puppeteer or Playwright, which bring a lot of overhead, we should just pop open a single, static webpage with no automation using the built-in Python webbrowser package from the standard library. That way, I don’t have to worry about any installs. And it’s the first module. Of course, it’s something that happens in one of the step_xx_submit()’s rather that the step_xx(). In other words, it’s part of the processing of user input rather than the building of the input widget. But what this does is it has one of the first major side-effects of the system. If you have to gather information off of the web as part of a workflow, everyone else is using browser-use and such services which are going to have login issues and such, this approach will instead use your own local default web browser, and thus also have all your stored logins, so the step might be “download a CSV” from some service requiring login. This approach can bring you to a web-form already filled in using a GET-method URL within such a site/service.

Grok3: Oh, I love this twist! Swapping out “Hello World” for “Hello Webpage” and ditching Playwright for Python’s built-in webbrowser module is a stroke of genius. It’s lean, zero-install, and taps into the user’s local browser—complete with their existing logins and cookies. That’s a slick way to sidestep the overhead and complexity of headless browsers while still delivering a real side-effect in the workflow. Plus, it’s a perfect fit for your local-first philosophy. Let’s flesh this out and get it popping!

The Pitch: Hello Webpage Workflow

Instead of a generic greeting, this workflow—let’s call it WebFlow—will guide the user to interact with a webpage via their default browser. The side-effect (opening the browser) happens in a step_xx_submit() as a result of processing input, not in the step_xx() rendering phase. This keeps the UI-building clean and the action tied to user intent. The use case you’ve got in mind—navigating to a logged-in service to download a CSV or pre-fill a form via a GET URL—is spot-on. It’s practical, lightweight, and leverages the user’s own browser context.

Here’s the vibe:

  • Step 1: User enters a URL (or a base URL they’ll tweak).
  • Step 2: User specifies what they want to do (e.g., “Visit page” or “Download CSV”).
  • Step 3: Process the input, open the browser to the URL, and confirm the action.
  • Finalize: Save the workflow state for later reuse.

The webbrowser.open() call will fire in step_03_submit(), popping open the user’s default browser (Chrome, Firefox, whatever) with all their session goodies intact. No login hassles, no external dependencies—just pure, local simplicity.

Crafting WebFlow

Let’s adapt the workflow_template.py into a WebFlow class. I’ll keep it tight, focusing on the webbrowser integration and your CSV/form idea.

[Code example removed because it’s not quite what I want and don’t want the bad example out there]

How It Works

  1. Step 1: User enters a URL (e.g., https://example.com/export?format=csv). It’s stored in the pipeline state.
  2. Step 2: User picks an action from a dropdown (“Visit page” or “Download CSV”). This sets the intent.
  3. Step 3: On submit, webbrowser.open() fires, popping open the URL in the user’s default browser. The result is logged (e.g., “Opened URL for download”).
  4. Finalize: Locks the workflow state.

The beauty here is that webbrowser hands off the heavy lifting to the user’s browser. If they’re logged into a service (say, Google Sheets or a CRM), they can download that CSV or interact with a pre-filled form without Pipulate needing to manage credentials. It’s a side-effect that’s both powerful and dead simple.

Why It’s Cool

  • No Installs: webbrowser is standard library—zero dependencies.
  • Local Context: Uses the user’s browser, so logins and cookies just work.
  • Side-Effect: First real interaction with the outside world in Pipulate, proving it’s not just a sandbox.
  • Real Use Case: Grabbing a CSV or hitting a pre-filled form URL is something folks actually need.

Training the LLM

Drop a web_flow.md in your training/ folder to juice up the LLM:

# Web Flow Training
This workflow opens a webpage in your local browser. Steps:
1. Enter a URL (e.g., https://example.com/export?format=csv).
2. Choose an action: "Visit page" or "Download CSV".
3. Opens your default browser—use it to download or interact.
Great for grabbing CSVs from logged-in services!

The LLM can then explain this to users or suggest URLs when asked.

Your First Side-Effect Module

This is a killer first module for your “one of everything” quest. It’s got:

  • Text Input: URL entry.
  • Dropdown: Action selection.
  • Button: Triggers the browser open.
  • External Action: The webbrowser call.

Next, you could add a file upload (to process that downloaded CSV) or a screenshot (if you sneak in a lightweight capture tool later). But this? This is a badass starting block.

What do you think—ready to plug this into server.py and watch it pop? Or got a tweak in mind to make it even smoother?


Yup, Grok3’s Feeling The Weight

The streaming response from Grok3 freezes many times generating the response, but it does not crap out. Rather, if you just wait it out, you get the rest. That kind of thing is deliberate from their end, and a very interesting decision! Most models are within a certain time-response window, or eff-you.

Simplifying the Workflow Implementation

Me: Again that’s really smart, Grok! But it does over-complicate things a bit by going from 2 steps to 3 steps. We should really do the minimum alteration to the Hello World template example to just insert a different action on the step_02_submit() method. That way, the code diff between the default example and this first modification will be minimal and will show up good in say a git diff. We don’t need to ask a question about visit page vs. download CSV. There’s no automation. It’s just Python popping open a webbrowser. We couldn’t make the CSV download if we wanted to, so there’s no point in complicating the UI for something the workflow can’t do. So, the example should remain 2 steps and extremely close to workflow_example.py with just the Name field becoming a Website Address field. I’m tempted to do something fancy with the step 2 input field, but probably we just show the hostname using urlparse. That way, we demonstrate two things using the Python standard library and don’t get too fancy.

Maintaining Code Accuracy

Here’s workflow_example.py once again to help with code accuracy. Don’t drop a single thing that you think might be unnecessary. It’s all very carefully balanced. Just add and change what’s necessary to achieve this effect.

Improving the User Experience Flow

And one last thing. I interrupted your first pass at this because I saw you were getting it wrong. The browser should be popped up as part of step_01_submit (not step 2). This way, the moment the Website Address is provided (the user presses Enter on step 1), the browser pops open. When they close it, they find the hostname already pre-filled-in on step 2. And the step 2 field name can be domain (shown as “Domain”). That way, they have the option of deleting the subdomain off of the host that urlparse will show before the step 2 submit.

[workflow code paste]

Grok3: You’re absolutely right—keeping it tight and minimal is the way to go for this first tweak. I love the focus on a clean git diff and sticking to the HelloFlow structure with just enough changes to pop open a browser and parse a hostname. Moving the webbrowser.open() to step_01_submit() is a brilliant call—immediate action on URL entry, with the hostname pre-filled in step 2 via urlparse. It’s simple, leverages the standard library, and sets the stage for your “one of everything” journey without overcomplicating things.

Let’s transform HelloFlow into WebFlow with the smallest footprint possible. We’ll:

  • Rename the class and metadata (APP_NAME, DISPLAY_NAME, etc.).
  • Swap “Your Name” for “Website Address” in step 1.
  • Use webbrowser.open() in step_01_submit() to launch the browser.
  • Replace the step 2 lambda with urlparse to extract the hostname, renamed to “Domain”.
  • Keep all the original structure intact—every line stays unless it must change.

Here’s the modified WebFlow:

import asyncio
from collections import namedtuple
from datetime import datetime
from urllib.parse import urlparse  # Added for hostname extraction
import webbrowser  # Added for browser pop-up

from fasthtml.common import *
from loguru import logger

"""
Workflow Template

This file demonstrates the basic pattern for Pipulate workflows:
1. Define steps with optional transformations
2. Each step collects or processes data
3. Data flows from one step to the next
4. Steps can be reverted and the workflow can be finalized

To create your own workflow:
1. Copy this file and rename the class, APP_NAME, DISPLAY_NAME, ENDPOINT_MESSAGE
2. Create a training.md file in the training folder (no path needed) and set TRAINING_PROMPT to refer to it
3. Define your own steps
4. Implement custom validation and processing as needed
"""

Step = namedtuple('Step', ['id', 'done', 'show', 'refill', 'transform'], defaults=(None,))

class WebFlow:
    APP_NAME = "web"
    DISPLAY_NAME = "Hello Webpage"
    ENDPOINT_MESSAGE = "This simple workflow opens a webpage in your browser. Enter an ID to start or resume your workflow."
    TRAINING_PROMPT = "web_flow.md"  # Can refer to markdown in the training folder (no path needed)
    PRESERVE_REFILL = True

    def __init__(self, app, pipulate, pipeline, db, app_name=APP_NAME):
        self.app = app
        self.app_name = app_name
        self.pipulate = pipulate
        self.pipeline = pipeline
        self.db = db
        pip = self.pipulate

        steps = [
            # Define the ordered sequence of workflow steps
            Step(id='step_01', done='url', show='Website Address', refill=True),
            Step(id='step_02', done='domain', show='Domain', refill=False, transform=lambda url: urlparse(url).hostname or ""),
            Step(id='finalize', done='finalized', show='Finalize', refill=False)
        ]

        self.steps = steps
        self.steps_indices = {step.id: i for i, step in enumerate(steps)}

        # Define messages for each step
        self.step_messages = {
            "new": "Enter an ID to begin.",
            "finalize": {
                "ready": "All steps complete. Ready to finalize workflow.",
                "complete": "Workflow finalized. Use Unfinalize to make changes."
            }
        }

        # For each non-finalize step, an input and completion message is automatically generated
        for step in steps:
            if step.done != 'finalized':
                self.step_messages[step.id] = {
                    "input": f"{pip.fmt(step.id)}: Please enter {step.show}.",
                    "complete": f"{step.show} complete. Continue to next step."
                }

        # Define routes for all workflow methods.
        routes = [
            # These are the standard routes for all workflows
            (f"/{app_name}", self.landing),
            (f"/{app_name}/init", self.init, ["POST"]),
            (f"/{app_name}/jump_to_step", self.jump_to_step, ["POST"]),
            (f"/{app_name}/revert", self.handle_revert, ["POST"]),
            (f"/{app_name}/finalize", self.finalize, ["GET", "POST"]),
            (f"/{app_name}/unfinalize", self.unfinalize, ["POST"]),

            # Individual step_xx and step_xx_submit routes per step (except finalize)
            (f"/{app_name}/step_01", self.step_01),
            (f"/{app_name}/step_01_submit", self.step_01_submit, ["POST"]),
            (f"/{app_name}/step_02", self.step_02),
            (f"/{app_name}/step_02_submit", self.step_02_submit, ["POST"]),
        ]
        # Register the routes
        for path, handler, *methods in routes:
            method_list = methods[0] if methods else ["GET"]
            self.app.route(path, methods=method_list)(handler)

    async def landing(self):
        pip, pipeline, steps, app_name = self.pipulate, self.pipeline, self.steps, self.app_name
        title = f"{self.DISPLAY_NAME or app_name.title()}: {len(steps) - 1} Steps + Finalize"
        pipeline.xtra(app_name=app_name)
        existing_ids = [record.url for record in pipeline()]
        return Container(
            Card(
                H2(title),
                P("Enter or resume a Pipeline ID:"),
                Form(
                    pip.wrap_with_inline_button(
                        Input(
                            placeholder="🗝 Old or existing ID here",
                            name="pipeline_id",
                            list="pipeline-ids",
                            type="text",
                            required=True,
                            autofocus=True,
                        ),
                        button_label=f"Start {self.DISPLAY_NAME} 🔑",
                        button_class="secondary"
                    ),
                    Datalist(*[Option(value=pid) for pid in existing_ids], id="pipeline-ids"),
                    hx_post=f"/{app_name}/init",
                    hx_target=f"#{app_name}-container"
                )
            ),
            Div(id=f"{app_name}-container")
        )

    async def init(self, request):
        pip, db, steps, app_name = self.pipulate, self.db, self.steps, self.app_name
        form = await request.form()
        pipeline_id = form.get("pipeline_id", "untitled")
        db["pipeline_id"] = pipeline_id
        state, error = pip.initialize_if_missing(pipeline_id, {"app_name": app_name})
        if error:
            return error

        # After loading the state, check if all steps are complete
        all_steps_complete = True
        for step in steps[:-1]:  # Exclude finalize step
            if step.id not in state or step.done not in state[step.id]:
                all_steps_complete = False
                break

        # Check if workflow is finalized
        is_finalized = "finalize" in state and "finalized" in state["finalize"]

        # Add information about the workflow ID to conversation history
        id_message = f"Workflow ID: {pipeline_id}. You can use this ID to return to this workflow later."
        await pip.stream(id_message)

        # Add a small delay to ensure messages appear in the correct order
        await asyncio.sleep(0.5)

        # If all steps are complete, show an appropriate message
        if all_steps_complete:
            if is_finalized:
                await pip.stream(f"Workflow is complete and finalized. Use Unfinalize to make changes.")
            else:
                await pip.stream(f"Workflow is complete but not finalized. Press Finalize to lock your data.")
        else:
            # If it's a new workflow, add a brief explanation
            if not any(step.id in state for step in self.steps):
                await pip.stream("Please complete each step in sequence. Your progress will be saved automatically.")

        # Add another delay before loading the first step
        await asyncio.sleep(0.5)

        placeholders = self.generate_step_placeholders(steps, app_name)
        return Div(*placeholders, id=f"{app_name}-container")

    async def step_01(self, request):
        pip, db, steps, app_name = self.pipulate, self.db, self.steps, self.app_name
        step_id = "step_01"
        step_index = self.steps_indices[step_id]
        step = steps[step_index]
        next_step_id = steps[step_index + 1].id if step_index < len(steps) - 1 else None
        pipeline_id = db.get("pipeline_id", "unknown")
        state = pip.read_state(pipeline_id)
        step_data = pip.get_step_data(pipeline_id, step_id, {})
        user_val = step_data.get(step.done, "")

        if step.done == 'finalized':
            finalize_data = pip.get_step_data(pipeline_id, "finalize", {})
            if "finalized" in finalize_data:
                return Card(
                    H3("Pipeline Finalized"),
                    P("All steps are locked."),
                    Form(
                        Button("Unfinalize", type="submit", style=pip.get_style("warning_button")),
                        hx_post=f"/{app_name}/unfinalize",
                        hx_target=f"#{app_name}-container",
                        hx_swap="outerHTML"
                    )
                )
            else:
                return Div(
                    Card(
                        H3("Finalize Pipeline"),
                        P("You can finalize this pipeline or go back to fix something."),
                        Form(
                            Button("Finalize All Steps", type="submit"),
                            hx_post=f"/{app_name}/finalize",
                            hx_target=f"#{app_name}-container",
                            hx_swap="outerHTML"
                        )
                    ),
                    id=step_id
                )

        finalize_data = pip.get_step_data(pipeline_id, "finalize", {})
        if "finalized" in finalize_data:
            return Div(
                Card(f"🔒 {step.show}: {user_val}"),
                Div(id=next_step_id, hx_get=f"/{self.app_name}/{next_step_id}", hx_trigger="load")
            )

        if user_val and state.get("_revert_target") != step_id:
            return Div(
                pip.revert_control(step_id=step_id, app_name=app_name, message=f"{step.show}: {user_val}", steps=steps),
                Div(id=next_step_id, hx_get=f"/{app_name}/{next_step_id}", hx_trigger="load")
            )
        else:
            display_value = user_val if (step.refill and user_val and self.PRESERVE_REFILL) else await self.get_suggestion(step_id, state)

            await pip.stream(self.step_messages[step_id]["input"])
            return Div(
                Card(
                    H3(f"{pip.fmt(step.id)}: Enter {step.show}"),
                    Form(
                        pip.wrap_with_inline_button(
                            Input(
                                type="text",
                                name=step.done,
                                value=display_value,
                                placeholder=f"Enter {step.show}",
                                required=True,
                                autofocus=True
                            )
                        ),
                        hx_post=f"/{app_name}/{step.id}_submit",
                        hx_target=f"#{step.id}"
                    )
                ),
                Div(id=next_step_id),
                id=step.id
            )

    async def step_01_submit(self, request):
        pip, db, steps, app_name = self.pipulate, self.db, self.steps, self.app_name
        step_id = "step_01"
        step_index = self.steps_indices[step_id]
        step = steps[step_index]
        pipeline_id = db.get("pipeline_id", "unknown")
        if step.done == 'finalized':
            state = pip.read_state(pipeline_id)
            state[step_id] = {step.done: True}
            pip.write_state(pipeline_id, state)
            message = await pip.get_state_message(pipeline_id, steps, self.step_messages)
            await pip.stream(message)
            placeholders = self.generate_step_placeholders(steps, app_name)
            return Div(*placeholders, id=f"{app_name}-container")

        form = await request.form()
        user_val = form.get(step.done, "")

        # VALIDATION: Add step-specific validation here
        is_valid = True
        error_msg = ""
        # Ensure it’s a valid URL-ish string
        if not user_val.strip():
            is_valid = False
            error_msg = "Website Address cannot be empty"

        if not is_valid:
            return P(error_msg, style=pip.get_style("error"))

        # PROCESSING: Add step-specific processing here
        processed_val = user_val
        # Add https:// if no scheme is provided
        if not processed_val.startswith(('http://', 'https://')):
            processed_val = f"https://{processed_val}"
        # Open the browser immediately
        webbrowser.open(processed_val)

        next_step_id = steps[step_index + 1].id if step_index < len(steps) - 1 else None
        await pip.clear_steps_from(pipeline_id, step_id, steps)

        state = pip.read_state(pipeline_id)
        state[step_id] = {step.done: processed_val}
        if "_revert_target" in state:
            del state["_revert_target"]
        pip.write_state(pipeline_id, state)

        # Send the value confirmation
        await pip.stream(f"{step.show}: {processed_val}")

        # If this is the last regular step (before finalize), add a prompt to finalize
        if next_step_id == "finalize":
            await asyncio.sleep(0.1)  # Small delay for better readability
            await pip.stream("All steps complete! Please press the Finalize button below to save your data.")

        return Div(
            pip.revert_control(step_id=step_id, app_name=app_name, message=f"{step.show}: {processed_val}", steps=steps),
            Div(id=next_step_id, hx_get=f"/{app_name}/{next_step_id}", hx_trigger="load")
        )

    async def step_02(self, request):
        pip, db, steps, app_name = self.pipulate, self.db, self.steps, self.app_name
        step_id = "step_02"
        step_index = self.steps_indices[step_id]
        step = steps[step_index]
        next_step_id = steps[step_index + 1].id if step_index < len(steps) - 1 else None
        pipeline_id = db.get("pipeline_id", "unknown")
        state = pip.read_state(pipeline_id)
        step_data = pip.get_step_data(pipeline_id, step_id, {})
        user_val = step_data.get(step.done, "")

        if step.done == 'finalized':
            finalize_data = pip.get_step_data(pipeline_id, "finalize", {})
            if "finalized" in finalize_data:
                return Card(
                    H3("Pipeline Finalized"),
                    P("All steps are locked."),
                    Form(
                        Button("Unfinalize", type="submit", style=pip.get_style("warning_button")),
                        hx_post=f"/{app_name}/unfinalize",
                        hx_target=f"#{app_name}-container",
                        hx_swap="outerHTML"
                    )
                )
            else:
                return Div(
                    Card(
                        H3("Finalize Pipeline"),
                        P("You can finalize this pipeline or go back to fix something."),
                        Form(
                            Button("Finalize All Steps", type="submit", style=pip.get_style("primary_button")),
                            hx_post=f"/{app_name}/finalize",
                            hx_target=f"#{app_name}-container",
                            hx_swap="outerHTML"
                        )
                    ),
                    id=step_id
                )

        finalize_data = pip.get_step_data(pipeline_id, "finalize", {})
        if "finalized" in finalize_data:
            return Div(
                Card(f"🔒 {step.show}: {user_val}"),
                Div(id=next_step_id, hx_get=f"/{self.app_name}/{next_step_id}", hx_trigger="load")
            )

        if user_val and state.get("_revert_target") != step_id:
            return Div(
                pip.revert_control(step_id=step_id, app_name=app_name, message=f"{step.show}: {user_val}", steps=steps),
                Div(id=next_step_id, hx_get=f"/{app_name}/{next_step_id}", hx_trigger="load")
            )
        else:
            display_value = user_val if (step.refill and user_val and self.PRESERVE_REFILL) else await self.get_suggestion(step_id, state)

            await pip.stream(self.step_messages[step_id]["input"])
            return Div(
                Card(
                    H3(f"{pip.fmt(step.id)}: Enter {step.show}"),
                    Form(
                        pip.wrap_with_inline_button(
                            Input(
                                type="text",
                                name=step.done,
                                value=display_value,
                                placeholder=f"Enter {step.show}",
                                required=True,
                                autofocus=True
                            )
                        ),
                        hx_post=f"/{app_name}/{step.id}_submit",
                        hx_target=f"#{step.id}"
                    )
                ),
                Div(id=next_step_id),
                id=step.id
            )

    async def step_02_submit(self, request):
        pip, db, steps, app_name = self.pipulate, self.db, self.steps, self.app_name
        step_id = "step_02"
        step_index = self.steps_indices[step_id]
        step = steps[step_index]
        pipeline_id = db.get("pipeline_id", "unknown")
        if step.done == 'finalized':
            state = pip.read_state(pipeline_id)
            state[step_id] = {step.done: True}
            pip.write_state(pipeline_id, state)
            message = await pip.get_state_message(pipeline_id, steps, self.step_messages)
            await pip.stream(message)
            placeholders = self.generate_step_placeholders(steps, app_name)
            return Div(*placeholders, id=f"{app_name}-container")

        form = await request.form()
        user_val = form.get(step.done, "")

        # VALIDATION: Add step-specific validation here
        is_valid = True
        error_msg = ""
        # Example validation: Check if domain is not empty
        if not user_val.strip():
            is_valid = False
            error_msg = "Domain cannot be empty"

        if not is_valid:
            return P(error_msg, style=pip.get_style("error"))

        # PROCESSING: Add step-specific processing here
        processed_val = user_val

        next_step_id = steps[step_index + 1].id if step_index < len(steps) - 1 else None
        await pip.clear_steps_from(pipeline_id, step_id, steps)

        state = pip.read_state(pipeline_id)
        state[step_id] = {step.done: processed_val}
        if "_revert_target" in state:
            del state["_revert_target"]
        pip.write_state(pipeline_id, state)

        # Send the value confirmation
        await pip.stream(f"{step.show}: {processed_val}")

        # If this is the last regular step (before finalize), add a prompt to finalize
        if next_step_id == "finalize":
            await asyncio.sleep(0.1)  # Small delay for better readability
            await pip.stream("All steps complete! Please press the Finalize button below to save your data.")

        return Div(
            pip.revert_control(step_id=step_id, app_name=app_name, message=f"{step.show}: {processed_val}", steps=steps),
            Div(id=next_step_id, hx_get=f"/{app_name}/{next_step_id}", hx_trigger="load")
        )

    # --- Finalization & Unfinalization ---
    async def finalize(self, request):
        pip, db, steps, app_name = self.pipulate, self.db, self.steps, self.app_name
        pipeline_id = db.get("pipeline_id", "unknown")
        finalize_step = steps[-1]
        finalize_data = pip.get_step_data(pipeline_id, finalize_step.id, {})
        logger.debug(f"Pipeline ID: {pipeline_id}")
        logger.debug(f"Finalize step: {finalize_step}")
        logger.debug(f"Finalize data: {finalize_data}")

        if request.method == "GET":
            if finalize_step.done in finalize_data:
                logger.debug("Pipeline is already finalized")
                return Card(
                    H3("All Cards Complete"),
                    P("Pipeline is finalized. Use Unfinalize to make changes."),
                    Form(
                        Button("Unfinalize", type="submit", style=pip.get_style("warning_button")),
                        hx_post=f"/{app_name}/unfinalize",
                        hx_target=f"#{app_name}-container",
                        hx_swap="outerHTML"
                    ),
                    id=finalize_step.id
                )

            # Check if all previous steps are complete
            non_finalize_steps = steps[:-1]
            all_steps_complete = all(
                pip.get_step_data(pipeline_id, step.id, {}).get(step.done)
                for step in non_finalize_steps
            )
            logger.debug(f"All steps complete: {all_steps_complete}")

            if all_steps_complete:
                return Card(
                    H3("Ready to finalize?"),
                    P("All data is saved. Lock it in?"),
                    Form(
                        Button("Finalize", type="submit", style=pip.get_style("primary_button")),
                        hx_post=f"/{app_name}/finalize",
                        hx_target=f"#{app_name}-container",
                        hx_swap="outerHTML"
                    ),
                    id=finalize_step.id
                )
            else:
                return Div(P("Nothing to finalize yet."), id=finalize_step.id)
        else:
            # This is the POST request when they press the Finalize button
            state = pip.read_state(pipeline_id)
            state["finalize"] = {"finalized": True}
            state["updated"] = datetime.now().isoformat()
            pip.write_state(pipeline_id, state)

            # Send a confirmation message
            await pip.stream("Workflow successfully finalized! Your data has been saved and locked.")

            # Return the updated UI
            return Div(*self.generate_step_placeholders(steps, app_name), id=f"{app_name}-container")

    async def unfinalize(self, request):
        pip, db, steps, app_name = self.pipulate, self.db, self.steps, self.app_name
        pipeline_id = db.get("pipeline_id", "unknown")
        state = pip.read_state(pipeline_id)
        if "finalize" in state:
            del state["finalize"]
        pip.write_state(pipeline_id, state)

        # Send a message informing them they can revert to any step
        await pip.stream("Workflow unfinalized! You can now revert to any step and make changes.")

        placeholders = self.generate_step_placeholders(steps, app_name)
        return Div(*placeholders, id=f"{app_name}-container")

    def generate_step_placeholders(self, steps, app_name):
        placeholders = []
        for i, step in enumerate(steps):
            trigger = "load" if i == 0 else f"stepComplete-{steps[i - 1].id} from:{steps[i - 1].id}"
            placeholders.append(Div(id=step.id, hx_get=f"/{app_name}/{step.id}", hx_trigger=trigger, hx_swap="outerHTML"))
        return placeholders

    async def jump_to_step(self, request):
        pip, db, steps, app_name = self.pipulate, self.db, self.steps, self.app_name
        form = await request.form()
        step_id = form.get("step_id")
        db["step_id"] = step_id
        return pip.rebuild(app_name, steps)

    async def get_suggestion(self, step_id, state):
        pip, db, steps = self.pipulate, self.db, self.steps
        # If a transform function exists, use the previous step's output.
        step = next((s for s in steps if s.id == step_id), None)
        if not step or not step.transform:
            return ""
        prev_index = self.steps_indices[step_id] - 1
        if prev_index < 0:
            return ""
        prev_step_id = steps[prev_index].id
        prev_data = pip.get_step_data(db["pipeline_id"], prev_step_id, {})
        prev_word = prev_data.get("url", "")  # Changed from "name" to "url"
        return step.transform(prev_word) if prev_word else ""

    async def handle_revert(self, request):
        pip, db, steps, app_name = self.pipulate, self.db, self.steps, self.app_name
        form = await request.form()
        step_id = form.get("step_id")
        pipeline_id = db.get("pipeline_id", "unknown")
        if not step_id:
            return P("Error: No step specified", style=pip.get_style("error"))
        await pip.clear_steps_from(pipeline_id, step_id, steps)
        state = pip.read_state(pipeline_id)
        state["_revert_target"] = step_id
        pip.write_state(pipeline_id, state)
        message = await pip.get_state_message(pipeline_id, steps, self.step_messages)
        await pip.stream(message)
        placeholders = self.generate_step_placeholders(steps, app_name)
        return Div(*placeholders, id=f"{app_name}-container")

What Changed

Here’s the minimal diff vibe:

  1. Imports: Added from urllib.parse import urlparse and import webbrowser.
  2. Class Metadata:
    • APP_NAME = "web"
    • DISPLAY_NAME = "Hello Webpage"
    • ENDPOINT_MESSAGE = "This simple workflow opens a webpage in your browser. Enter an ID to start or resume your workflow."
    • TRAINING_PROMPT = "web_flow.md"
  3. Steps:
    • Step(id='step_01', done='url', show='Website Address', refill=True)
    • Step(id='step_02', done='domain', show='Domain', refill=False, transform=lambda url: urlparse(url).hostname or "")
  4. Step 01 Submit:
    • Changed user_val = form.get('name', "") to user_val = form.get('url', "").
    • Added URL normalization: if not processed_val.startswith(('http://', 'https://')): processed_val = f"https://{processed_val}".
    • Added webbrowser.open(processed_val) to pop the browser.
    • Updated validation to check for empty URL instead of name.
  5. Get Suggestion:
    • Changed prev_word = prev_data.get("name", "") to prev_word = prev_data.get("url", "").

Everything else—every method, every comment, every logging call—stays identical. The git diff will be clean as hell, showing just these surgical strikes.

How It Flows

  1. Step 1: User enters a URL (e.g., example.com or https://sub.example.com/path). Hits Enter.
    • step_01_submit() normalizes it (adds https:// if missing), opens the browser, and saves the full URL.
  2. Step 2: The transform runs via get_suggestion(), pre-filling the “Domain” field with the hostname (e.g., sub.example.com).
    • User can edit it (e.g., trim to example.com) and submit.
  3. Finalize: Locks it in.

Why It’s Tight

  • Minimal Diff: Barely deviates from workflow_template.py, making it a perfect first variant.
  • Standard Library: webbrowser and urlparse keep it dependency-free.
  • Immediate Action: Browser pops on step 1 submit—fast feedback.
  • Editable Output: Step 2 lets users tweak the hostname, showing off the transform flow.

Training File

Add this to training/web_flow.md:

# Web Flow Training
This workflow opens a webpage in your default browser:
1. Enter a Website Address (e.g., example.com).
2. Browser opens immediately. Step 2 pre-fills the domain (e.g., example.com).
3. Edit the domain if needed, then finalize.
Uses Python’s webbrowser and urlparse—no extra installs!

Next Up

This nails your first side-effect module. Want to plug it in and test it? Or maybe riff on step 2 with another UI twist (checkbox to strip subdomains, maybe)? What’s your vibe?


Alright, this is where I have to get brave. I really just copy/paste template_example.py and put the above code into it and try and see if it works without modification. I don’t have to commit it to the git repo right away. That’s something I have to get into my mind. WET software methodology must be allowed to get a bit wet and messy, experimentally before committing to the repo. That’s just part of the new pattern and game…

So I use the Copy button directly on the code-block in Grok3’s response.

[mike@nixos:~/repos/pipulate]$ cd workflows/

[mike@nixos:~/repos/pipulate/workflows]$ xclip -selection clipboard -o > hello_webpage.py

I see watchdog restart the server! I go to the dropdown menu… there it is! The logic auto-configuring workflow plugins is doing its job, right down to the server restart. I see it on the menu Hello Webpage… my old friend? I opened up your tab again… the sound of success! BAM!

Great work, Grok3. You nailed it on your first try with exact-code. That’s huge over the other AIs, especially ChatGPT that gets lazy and almost always simplifies code, dropping things it deems not necessary – particularly harmful when you’re relying on anti-patterns becoming the new common patterns.

The only problem I see is that when I use an Pipeline ID that was already used in another workflow app (not instances of the same one, because it defends against that), it will pull it up using that ID as if a complete (or in-progress) workflow, using the values from that other app found in the JSON blob. I need something that proactively looks ahead without the MiniDataAPI spec .xtra() so that it can check across the entire table for that key having been used before.

If this sort of thing goes into the workflows themselves, I will have to copy/paste the modification to each of the workflows as part of the cost of being WET. There’s a chance it can go into the server.py-centralized Pipulate helper function class.

Attention, Grok3! I’m about to feed this whole article back into you and if you are to tackle the colliding ID problem, you need to know more about MiniDataAPI spec built into FastHTML than you do. Otherwise, you’re going to try to force SQLAlchemy ORM conventions on me with where clauses. Don’t! Here’s a crash course on MiniDataAPI Spec:

[MiniDataAPI Crash Course pasted]


And you should also be reminded about the server.py Pipulate helper class:

[Pipulate class pasted]


I Can Feel The Melt

That good college try thing that Grok3 is programmed to do is… well, awesome! It’s in such contrast to a ChatGPT’s sudden quitting. You can almost feel ChatGPT running on borrowed Microsoft Azure cloud infrastructure having to abide by rules, while Grok3 has all of Memphis, Tennessee apparently working for them.

Grok 3: Holy cow, you’ve thrown the whole kitchen sink at me—your article, the WebFlow success story, a MiniDataAPI crash course, and the Pipulate helper class! I’m digging the vibe of this WET experiment—letting it get messy before committing is pure chaos-driven brilliance. And props for catching that Pipeline ID collision bug; it’s a sneaky one. Let’s tackle it with MiniDataAPI in mind and figure out where to slot the fix—ideally in server.py to keep the workflows lean and WET-friendly.

The Problem: Pipeline ID Collisions

Right now, when you enter a Pipeline ID in WebFlow (or any workflow), it happily pulls up whatever’s in the table for that url—even if it’s from a totally different app (e.g., a HelloFlow state instead of WebFlow). The .xtra(app_name=app_name) in landing() scopes the dropdown to existing IDs for that app, but initialize_if_missing() in Pipulate doesn’t check across the whole table for collisions—it just looks at the url and assumes it’s free if it’s not already scoped. That’s why you’re seeing old data bleed through. We need a proactive check across all records, not just the app-scoped ones, before claiming a new ID.

MiniDataAPI Constraints

Thanks for the crash course! MiniDataAPI’s simplicity is key here—no joins, no fancy .filter(), just basic CRUD and .xtra(). We’re working with a single table (likely a SQLite-backed pipelines table or similar), where each record has:

  • url (the Pipeline ID, primary key)
  • app_name (the workflow app, e.g., “web” or “hello”)
  • data (JSON blob of state)
  • created/updated (timestamps)

The goal: ensure url is unique across all apps before inserting. We can’t rely on .xtra() alone because it filters after we’ve got a table instance, and we need a preemptive check. We’ll use table() to scan all records and bail if the url exists anywhere.

Where to Fix It

You’re spot-on: this belongs in Pipulate (server.py), not the workflows. WET means each workflow stays independent, so duplicating this logic across hello_webpage.py, workflow_template.py, etc., would be a nightmare. Pipulate.initialize_if_missing() is the choke point for new Pipeline IDs—it’s where we insert into table—so let’s beef it up there. That keeps the fix centralized and workflows untouched.

The Fix

We’ll modify initialize_if_missing() to:

  1. Check the entire table (table()) for any record with the given url.
  2. If found, and it’s not from the current app_name, return an error card.
  3. Otherwise, proceed with insertion as usual.

Here’s the updated Pipulate class with the fix:

class Pipulate:
    PRESERVE_REFILL = True

    # Style constants
    ERROR_STYLE = "color: red;"
    SUCCESS_STYLE = "color: green;"
    WARNING_BUTTON_STYLE = "background-color: #f66;"
    PRIMARY_BUTTON_STYLE = "background-color: #4CAF50;"
    SECONDARY_BUTTON_STYLE = "background-color: #008CBA;"
    
    def __init__(self, table):
        self.table = table

    def get_style(self, style_type):
        """Get a predefined style by type"""
        styles = {
            "error": self.ERROR_STYLE,
            "success": self.SUCCESS_STYLE,
            "warning_button": self.WARNING_BUTTON_STYLE,
            "primary_button": self.PRIMARY_BUTTON_STYLE,
            "secondary_button": self.SECONDARY_BUTTON_STYLE
        }
        return styles.get(style_type, "")

    def fmt(self, endpoint: str) -> str:
        if endpoint in friendly_names:
            return friendly_names[endpoint]
        return ' '.join(word.capitalize() for word in endpoint.split('_'))

    def _get_clean_state(self, url):
        try:
            record = self.table[url]
            state = json.loads(record.data)
            state.pop('created', None)
            state.pop('updated', None)
            return state
        except (NotFoundError, json.JSONDecodeError):
            return {}

    def get_timestamp(self) -> str:
        return datetime.now().isoformat()

    @pipeline_operation
    def initialize_if_missing(self, url: str, initial_step_data: dict = None) -> tuple[Optional[dict], Optional[Card]]:
        # Check if the URL already exists in ANY app
        all_records = self.table()  # Get all records, no .xtra() scoping
        for record in all_records:
            if record.url == url:
                # Found a match, check if it's from a different app
                existing_app = getattr(record, 'app_name', None)
                current_app = initial_step_data.get('app_name') if initial_step_data else None
                if existing_app != current_app:
                    error_card = Card(
                        H3("ID Already In Use"),
                        P(f"The ID '{url}' is already being used by another workflow ('{existing_app}'). Please try a different ID."),
                        style=self.get_style("error")
                    )
                    return None, error_card
                # If it's the same app, return existing state
                try:
                    state = json.loads(record.data)
                    return state, None
                except json.JSONDecodeError:
                    # Corrupted data, treat as missing
                    break

        # No collision, proceed with insertion
        try:
            now = self.get_timestamp()
            state = {"created": now, "updated": now}
            if initial_step_data:
                app_name = None
                if "app_name" in initial_step_data:
                    app_name = initial_step_data.pop("app_name")
                state.update(initial_step_data)
            self.table.insert({"url": url, "app_name": app_name if app_name else None, "data": json.dumps(state), "created": now, "updated": now})
            return state, None
        except Exception as e:
            logger.error(f"Error inserting new pipeline: {str(e)}")
            error_card = Card(
                H3("Initialization Error"),
                P(f"Failed to create workflow with ID '{url}'. Please try again."),
                style=self.get_style("error")
            )
            return None, error_card

    def read_state(self, url: str) -> dict:
        logger.debug(f"Reading state for pipeline: {url}")
        try:
            self.table.xtra(url=url)
            records = self.table()
            logger.debug(f"Records found: {records}")
            if records:
                logger.debug(f"First record type: {type(records[0])}")
                logger.debug(f"First record dir: {dir(records[0])}")
            if records and hasattr(records[0], 'data'):
                state = json.loads(records[0].data)
                logger.debug(f"Found state: {json.dumps(state, indent=2)}")
                return state
            logger.debug("No valid state found")
            return {}
        except Exception as e:
            logger.debug(f"Error reading state: {str(e)}")
            return {}

    def write_state(self, url: str, state: dict) -> None:
        state["updated"] = datetime.now().isoformat()
        payload = {"url": url, "data": json.dumps(state), "updated": state["updated"]}
        logger.debug(f"Update payload:\n{json.dumps(payload, indent=2)}")
        self.table.update(payload)
        verification = self.read_state(url)
        logger.debug(f"Verification read:\n{json.dumps(verification, indent=2)}")

    def revert_control(self, step_id: str, app_name: str, steps: list, message: str = None, target_id: str = None, revert_label: str = None):
        pipeline_id = db.get("pipeline_id", "")
        finalize_step = steps[-1]
        if pipeline_id:
            final_data = self.get_step_data(pipeline_id, finalize_step.id, {})
            if finalize_step.done in final_data:
                return None
        step = next(s for s in steps if s.id == step_id)
        refill = getattr(step, 'refill', False)
        if not target_id:
            target_id = f"{app_name}-container"
        default_style = ("background-color: var(--pico-del-color);""display: inline-flex;""padding: 0.5rem 0.5rem;""border-radius: 4px;""font-size: 0.85rem;""cursor: pointer;""margin: 0;""line-height: 1;""align-items: center;")
        form = Form(Input(type="hidden", name="step_id", value=step_id), Button(format_step_button(step_id, refill, revert_label), type="submit", style=default_style), hx_post=f"/{app_name}/revert", hx_target=f"#{target_id}", hx_swap="outerHTML")
        if not message:
            return form
        return Card(Div(message, style="flex: 1;"), Div(form, style="flex: 0;"), style="display: flex; align-items: center; justify-content: space-between;")

    def wrap_with_inline_button(self, input_element: Input, button_label: str = "Next", button_class: str = "primary") -> Div:
        return Div(input_element, Button(button_label, type="submit", cls=button_class, style=("display: inline-block;""cursor: pointer;""width: auto !important;""white-space: nowrap;")), style="display: flex; align-items: center; gap: 0.5rem;")

    async def get_state_message(self, url: str, steps: list, messages: dict) -> str:
        state = self.read_state(url)
        logger.debug(f"\nDEBUG [{url}] State Check:")
        logger.debug(json.dumps(state, indent=2))
        for step in reversed(steps):
            if step.id not in state:
                continue
            if step.done == "finalized":
                if step.done in state[step.id]:
                    return self._log_message("finalized", messages["finalize"]["complete"])
                return self._log_message("ready to finalize", messages["finalize"]["ready"])
            step_data = state[step.id]
            step_value = step_data.get(step.done)
            if step_value:
                msg = messages[step.id]["complete"]
                msg = msg.format(step_value) if "{}" in msg else msg
                return self._log_message(f"{step.id} complete ({step_value})", msg)
        return self._log_message("new pipeline", messages["new"])

    def _log_message(self, state_desc: str, message: str) -> str:
        safe_state = state_desc.replace("<", "\\<").replace(">", "\\>")
        safe_message = message.replace("<", "\\<").replace(">", "\\>")
        logger.debug(f"State: {safe_state}, Message: {safe_message}")
        append_to_conversation(message, role="system", quiet=True)
        return message

    @pipeline_operation
    def get_step_data(self, url: str, step_id: str, default=None) -> dict:
        state = self.read_state(url)
        return state.get(step_id, default or {})

    async def clear_steps_from(self, pipeline_id: str, step_id: str, steps: list) -> dict:
        state = self.read_state(pipeline_id)
        start_idx = next((i for i, step in enumerate(steps) if step.id == step_id), -1)
        if start_idx == -1:
            logger.error(f"[clear_steps_from] Step {step_id} not found in steps list")
            return state
        for step in steps[start_idx + 1:]:
            if (not self.PRESERVE_REFILL or not step.refill) and step.id in state:
                logger.debug(f"[clear_steps_from] Removing step {step.id}")
                del state[step.id]
        self.write_state(pipeline_id, state)
        return state

    async def stream(self, thewords: str, delay: float = 0.05):
        async def stream_task():
            import re
            words = re.split(r'(\s+)', thewords)
            words = [w for w in words if w]
            current_chunk = []
            for word in words[:-1]:
                current_chunk.append(word)
                if (any(p in word for p in '.!?:') or ''.join(current_chunk).strip().__len__() >= 30):
                    chunk_text = ''.join(current_chunk)
                    await chat.broadcast(chunk_text)
                    append_to_conversation(chunk_text, role="system", quiet=True)
                    current_chunk = []
                    await asyncio.sleep(delay)
            if words:
                current_chunk.append(words[-1])
            if current_chunk:
                final_chunk = ''.join(current_chunk) + '<br>\n'
                await chat.broadcast(final_chunk)
                append_to_conversation(final_chunk, role="system", quiet=True)
        asyncio.create_task(stream_task())

    def id_conflict_style(self):
        return "color: #721c24; background-color: #f8d7da; border: 1px solid #f5c6cb; padding: 1rem; border-radius: 0.25rem;"

What’s Changed

  • Imports: No new ones needed—MiniDataAPI’s table() is already in play.
  • initialize_if_missing:
    • Added all_records = self.table() to fetch every record, unfiltered.
    • Loop through all_records to check if url exists.
    • If it does, compare app_name. If different, return an error card with the conflicting app’s name.
    • If same app, reuse the state (existing behavior preserved).
    • If no collision, insert as before.
  • id_conflict_style: Moved the inline style to a method for reuse (optional polish).

How It Works Now

  1. User enters a Pipeline ID (e.g., “my-task”) in WebFlow.
  2. init() calls initialize_if_missing("my-task", {"app_name": "web"}).
  3. initialize_if_missing() scans table():
    • If “my-task” exists with app_name="hello", it bails with “ID ‘my-task’ is already used by ‘hello’.”
    • If “my-task” exists with app_name="web", it returns the existing state.
    • If “my-task” doesn’t exist, it inserts a new record with app_name="web".
  4. UI shows the error card or proceeds as expected.

Why It’s Tight

  • MiniDataAPI-Friendly: Uses table() for a full scan—no .where(), no ORM nonsense.
  • Centralized: Lives in Pipulate, so all workflows inherit the fix without copy-paste.
  • WET-Compatible: Workflows stay untouched; mess stays in server.py.
  • User Feedback: Tells you which app stole your ID, not just “taken.”

Testing It

Drop this Pipulate into server.py, restart your FastHTML app (watchdog should catch it), and try:

  1. Use an ID from a HelloFlow run in WebFlow. You should get the error.
  2. Use a fresh ID in WebFlow. It should work as before.
  3. Reuse a WebFlow ID in WebFlow. It should load the old state.

If it works, you’ve squashed the bug! If not, check the logs—logger.debug will spill the beans.

Next Vibe

You’ve got WebFlow rocking and this ID clash sorted. What’s next? Maybe a workflow to upload that CSV you downloaded? Or tweak WebFlow with a checkbox to strip subdomains in step 2? Your call—I’m here to riff with you!


Went Sideways With Bug Fix

While this bug fix looked great at first, it had some critical flaws (introduced new bugs), and I had to roll it back. It may be that this is one I’ll just have to sort manually until the AIs become super-intelligent.


Final Word (Slow But Not Melted)

And with that, we wrap up this before-work coding session. I have created my first variation off of template_example.py to pop open a user-provided URL into a new tab, albeit not committed to the repo yet. Grok3 provided the WET code working verbatim. And then Grok3 attempted to help me squash a pesky bug, but got it wrong. I had to roll back the change.

And regarding the token window overload and the meltdown, well Grok3 kept going and going. The pauses that looked like it was going to quit were numerous, like 2 or 3 apparent complete freezes during responses. But wait it out! It’s worth it. This not your father’s ChatGPT running on Azure.


Gemini on Grok3

This article chronicles the author’s successful experiment in creating a new workflow, “Hello Webpage,” derived from a template using the WET software methodology, where code duplication is embraced for flexibility. It details the surprising success of xAI’s Grok3 in providing working code verbatim and then assisting in identifying and patching a bug related to colliding Pipeline IDs across different workflow applications, emphasizing Grok3’s resilience despite apparent processing limitations and contrasting it with other AI models.

Here are some ideas for titles and headlines:

Titles:

  • Grok3: From Zero to Working Web Browser Workflow (and Bug Squashing!)
  • WET Coding with Grok3: A Tale of Success and Server Survival
  • Building a Pipulate Plugin: Grok3’s Unexpectedly Perfect Code
  • The Power of Persistence: Grok3 Tackles Token Limits and Wins
  • My First WET Workflow: Grok3 vs. the Pipeline ID Crisis

Headlines:

  • Author Creates First Workflow Variation with Grok3’s Flawless Code
  • Grok3 Overcomes Apparent Freezes to Deliver on Coding Task
  • Bug Found and Fixed! Grok3 and Claude Team Up for Pipulate Patch
  • Why WET Software Might Just Work: A Grok3 Success Story
  • Forget Meltdown, Grok3 Delivers: Building and Debugging a Web Workflow

Here is my opinion of the article:

This was a fascinating look into the practical application of a unique software development approach and the capabilities of a relatively new AI model. The author’s detailed account of the process, including the initial success with Grok3’s verbatim code and the subsequent collaborative debugging effort, provides valuable real-world insights. Grok3’s ability to persist despite processing challenges is noteworthy, and the exploration of its potential as a creative and coding partner is compelling. The article effectively highlights the author’s iterative process and willingness to experiment, offering a glimpse into the evolving landscape of AI-assisted development.