The first few times I used an AI assistant for programming, the result felt almost unfair. A task that would normally take an hour appeared in less than a minute. There were classes, validation, logging and tests. The code looked calm and confident, which is probably the most dangerous visual style code can have.

Bad human code often warns you. Variable names are strange, methods are too long, and one comment says temporary fix from 2022. AI-generated code usually gives no such warning. It looks finished even when the model has only guessed what finished should look like.

So I started doing a small experiment. Instead of asking whether the generated code worked, I asked a different question: under what conditions does it stop being correct?

That changed everything.

Most failures were not syntax errors. They lived between functions, requests and database operations. One line was correct. The next line was correct. The complete operation was wrong.

Have you ever reviewed a pull request and felt that everything looked reasonable, but something still bothered you? AI code produces that feeling surprisingly often. The trick is learning where to look.

The happy path is where AI looks strongest

Consider a basic money transfer endpoint. The task sounds simple:

load two accounts
check the balance
subtract money from one account
add it to another
save both records

An AI assistant can generate this in seconds. It can also add input validation, custom exceptions and unit tests. Run one request and the balance changes correctly. Run another request after it and everything is still fine.

The bug appears when two transfers read the same balance before either of them writes the new value.

This is not a rare or theoretical situation. It can happen when a user double-clicks a button, when a mobile client retries after a timeout, or when two workers process related jobs. The generated code does not need to be obviously bad. It only needs to assume that the world waits politely between two lines.

Here is a simplified version of the first implementation.

Java — a transfer service that looks correct but loses consistency

package example.bank;

import java.math.BigDecimal;
import java.time.Instant;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;

public final class UnsafeTransferService {

    private final AccountRepository accountRepository;
    private final TransferRepository transferRepository;

    public UnsafeTransferService(
            AccountRepository accountRepository,
            TransferRepository transferRepository
    ) {
        this.accountRepository = accountRepository;
        this.transferRepository = transferRepository;
    }

    public TransferResult transfer(TransferCommand command) {
        validate(command);

        Account source = accountRepository.findById(command.sourceAccountId())
                .orElseThrow(() -> new IllegalArgumentException(
                        "Source account was not found"
                ));

        Account target = accountRepository.findById(command.targetAccountId())
                .orElseThrow(() -> new IllegalArgumentException(
                        "Target account was not found"
                ));

        if (source.balance().compareTo(command.amount()) < 0) {
            throw new IllegalStateException("Insufficient funds");
        }

        Account updatedSource = source.withBalance(
                source.balance().subtract(command.amount())
        );

        Account updatedTarget = target.withBalance(
                target.balance().add(command.amount())
        );

        accountRepository.save(updatedSource);
        accountRepository.save(updatedTarget);

        Transfer transfer = new Transfer(
                UUID.randomUUID(),
                command.sourceAccountId(),
                command.targetAccountId(),
                command.amount(),
                Instant.now()
        );

        transferRepository.save(transfer);

        return new TransferResult(
                transfer.id(),
                updatedSource.balance(),
                updatedTarget.balance()
        );
    }

    private static void validate(TransferCommand command) {
        if (command == null) {
            throw new IllegalArgumentException("Command is required");
        }

        if (command.sourceAccountId() == null ||
                command.targetAccountId() == null) {
            throw new IllegalArgumentException("Account IDs are required");
        }

        if (command.sourceAccountId().equals(command.targetAccountId())) {
            throw new IllegalArgumentException(
                    "Source and target must be different"
            );
        }

        if (command.amount() == null ||
                command.amount().compareTo(BigDecimal.ZERO) <= 0) {
            throw new IllegalArgumentException(
                    "Amount must be greater than zero"
            );
        }
    }

    public record TransferCommand(
            UUID sourceAccountId,
            UUID targetAccountId,
            BigDecimal amount
    ) {
    }

    public record TransferResult(
            UUID transferId,
            BigDecimal sourceBalance,
            BigDecimal targetBalance
    ) {
    }

    public record Transfer(
            UUID id,
            UUID sourceAccountId,
            UUID targetAccountId,
            BigDecimal amount,
            Instant createdAt
    ) {
    }

    public record Account(
            UUID id,
            BigDecimal balance,
            long version
    ) {
        public Account withBalance(BigDecimal newBalance) {
            return new Account(id, newBalance, version);
        }
    }

    public interface AccountRepository {
        java.util.Optional<Account> findById(UUID id);

        void save(Account account);
    }

    public interface TransferRepository {
        void save(Transfer transfer);
    }

    /*
     * This repository is only here to make the example executable.
     * ConcurrentHashMap makes individual operations thread-safe,
     * but it does not make the complete read-modify-write sequence atomic.
     */
    public static final class InMemoryAccountRepository
            implements AccountRepository {

        private final Map<UUID, Account> accounts =
                new ConcurrentHashMap<>();

        public void add(Account account) {
            accounts.put(account.id(), account);
        }

        @Override
        public java.util.Optional<Account> findById(UUID id) {
            return java.util.Optional.ofNullable(accounts.get(id));
        }

        @Override
        public void save(Account account) {
            accounts.put(account.id(), account);
        }
    }
}

Nothing here screams broken. Even ConcurrentHashMap is present, which creates a nice feeling of thread safety. But only separate map operations are thread-safe. The business operation is still a read-modify-write sequence spread across several calls.

Suppose the source account contains 100 units. Two requests try to transfer 80 units at nearly the same time. Both load the balance of 100. Both pass the check. Both calculate a remaining balance of 20. Depending on the database and update strategy, the system may record two successful transfers while subtracting the money only once.

That is not a formatting issue. It is an invariant failure.

The invariant should be stronger than the code:

The total amount of money must not change, and an account must never spend the same balance twice.

An AI assistant usually generates methods. Production systems need invariants.

A transaction annotation is not a magic shield

The obvious fix is to place the operation inside a database transaction. AI tools suggest this quickly, and sometimes that is enough. But a transaction alone does not prevent two transactions from reading the same old value.

The database isolation level matters. The update query matters. Locking matters. The order in which accounts are locked also matters, because careless locking can create deadlocks.

A safer design can use pessimistic row locking, optimistic version checks or a conditional atomic update. Each option has trade-offs. The important part is that the concurrency strategy must be explicit.

For example, an atomic debit can be expressed as a database operation that succeeds only when the current balance is large enough:

UPDATE accounts
SET balance = balance - :amount,
    version = version + 1
WHERE id = :account_id
  AND balance >= :amount;

Then the application checks the number of affected rows. Zero rows means the account does not exist or lacks funds. There is no gap between reading the balance and updating it.

This is the kind of detail AI-generated code often misses because it depends on a condition outside the method body. The prompt says create a transfer service. It rarely says preserve monetary invariants under concurrent execution, define a locking order, reject duplicate commands and remain correct after a process crash.

Should all of that be written in every prompt? Maybe. But at some point the prompt becomes a design document, and writing it correctly requires the same engineering knowledge needed to review the result.

That is the uncomfortable part. AI reduces typing much faster than it reduces responsibility.

Retries turn successful code into duplicate work

The second experiment involved webhooks. The service receives a payment event and activates a subscription. Again, the first generated version was neat:

parse JSON
verify the event type
find the user
activate the subscription
send an email
return HTTP 200

It worked perfectly in a local test.

Then the same event was sent twice.

Many external systems retry requests when they do not receive a successful response quickly enough. Your service may finish the database update but crash before sending the response. From the sender’s point of view, the request failed. From your database’s point of view, it succeeded.

The sender retries. Now the same logical event enters the system again.

This creates one of my favourite AI-code bugs: the code handles a request correctly but does not handle the fact that the request may have already been handled correctly.

The result can be duplicate emails, duplicated credits, repeated inventory updates or several background jobs for one event. Everything depends on what the handler does after receiving the message.

Idempotency is not a small improvement here. It is part of correctness.

Tests can confirm the same wrong assumption as the code

Generated tests deserve special suspicion. They often look impressive because they contain mocks, fixtures and descriptive method names. But many of them test the exact implementation the model just generated.

If the code assumes one request at a time, the tests also send one request at a time.

If the code forgets rollback behaviour, the tests mock repositories that never fail.

If the code trusts a timestamp, the tests use a timestamp that is always valid.

A green test suite can therefore mean only that the code and tests share the same blind spot.

For the webhook handler, I replaced ordinary unit tests with a small integration-style test. It sent the same event many times in parallel, inserted random failures after the database write and checked the final state.

The first version failed quickly.

Below is a safer implementation. It stores incoming event IDs, claims each event atomically and separates database changes from external side effects.

Python — an idempotent webhook handler with an outbox

from __future__ import annotations

import hashlib
import hmac
import json
import sqlite3
import time
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any


@dataclass(frozen=True)
class WebhookRequest:
    raw_body: bytes
    signature: str
    received_at: datetime


class InvalidSignatureError(Exception):
    pass


class InvalidPayloadError(Exception):
    pass


class WebhookProcessor:
    def __init__(
        self,
        connection: sqlite3.Connection,
        signing_secret: bytes,
        max_clock_skew_seconds: int = 300,
    ) -> None:
        self.connection = connection
        self.signing_secret = signing_secret
        self.max_clock_skew_seconds = max_clock_skew_seconds

    def process(self, request: WebhookRequest) -> str:
        payload = self._decode_and_verify(request)

        event_id = self._required_string(payload, 'id')
        event_type = self._required_string(payload, 'type')
        event_created_at = self._required_integer(payload, 'created_at')

        self._validate_event_time(
            event_created_at=event_created_at,
            received_at=request.received_at,
        )

        cursor = self.connection.cursor()

        try:
            cursor.execute('BEGIN IMMEDIATE')

            claimed = self._claim_event(
                cursor=cursor,
                event_id=event_id,
                event_type=event_type,
                raw_body=request.raw_body,
            )

            if not claimed:
                self.connection.rollback()
                return 'duplicate'

            if event_type == 'payment.completed':
                self._handle_payment_completed(cursor, payload)
            elif event_type == 'payment.refunded':
                self._handle_payment_refunded(cursor, payload)
            else:
                self._mark_event_ignored(cursor, event_id)

            self._mark_event_processed(cursor, event_id)
            self.connection.commit()
            return 'processed'

        except Exception:
            self.connection.rollback()
            raise

    def _decode_and_verify(
        self,
        request: WebhookRequest,
    ) -> dict[str, Any]:
        expected_signature = hmac.new(
            self.signing_secret,
            request.raw_body,
            hashlib.sha256,
        ).hexdigest()

        if not hmac.compare_digest(
            expected_signature,
            request.signature,
        ):
            raise InvalidSignatureError('Signature mismatch')

        try:
            payload = json.loads(request.raw_body)
        except json.JSONDecodeError as error:
            raise InvalidPayloadError('Invalid JSON') from error

        if not isinstance(payload, dict):
            raise InvalidPayloadError('Payload must be an object')

        return payload

    def _claim_event(
        self,
        cursor: sqlite3.Cursor,
        event_id: str,
        event_type: str,
        raw_body: bytes,
    ) -> bool:
        cursor.execute(
            '''
            INSERT INTO webhook_events (
                event_id,
                event_type,
                payload,
                status,
                created_at
            )
            VALUES (?, ?, ?, 'processing', CURRENT_TIMESTAMP)
            ON CONFLICT(event_id) DO NOTHING
            ''',
            (event_id, event_type, raw_body),
        )

        return cursor.rowcount == 1

    def _handle_payment_completed(
        self,
        cursor: sqlite3.Cursor,
        payload: dict[str, Any],
    ) -> None:
        data = payload.get('data')

        if not isinstance(data, dict):
            raise InvalidPayloadError('Missing event data')

        payment_id = self._required_string(data, 'payment_id')
        user_id = self._required_string(data, 'user_id')
        plan = self._required_string(data, 'plan')

        cursor.execute(
            '''
            INSERT INTO payments (
                payment_id,
                user_id,
                plan,
                status,
                created_at
            )
            VALUES (?, ?, ?, 'completed', CURRENT_TIMESTAMP)
            ON CONFLICT(payment_id) DO UPDATE SET
                status = excluded.status,
                plan = excluded.plan
            ''',
            (payment_id, user_id, plan),
        )

        cursor.execute(
            '''
            INSERT INTO subscriptions (
                user_id,
                plan,
                status,
                updated_at
            )
            VALUES (?, ?, 'active', CURRENT_TIMESTAMP)
            ON CONFLICT(user_id) DO UPDATE SET
                plan = excluded.plan,
                status = 'active',
                updated_at = CURRENT_TIMESTAMP
            ''',
            (user_id, plan),
        )

        cursor.execute(
            '''
            INSERT INTO outbox (
                event_key,
                event_type,
                payload,
                status,
                created_at
            )
            VALUES (?, 'subscription.activated', ?, 'pending',
                    CURRENT_TIMESTAMP)
            ON CONFLICT(event_key) DO NOTHING
            ''',
            (
                f'subscription-activated:{payment_id}',
                json.dumps(
                    {
                        'user_id': user_id,
                        'payment_id': payment_id,
                        'plan': plan,
                    }
                ),
            ),
        )

    def _handle_payment_refunded(
        self,
        cursor: sqlite3.Cursor,
        payload: dict[str, Any],
    ) -> None:
        data = payload.get('data')

        if not isinstance(data, dict):
            raise InvalidPayloadError('Missing event data')

        payment_id = self._required_string(data, 'payment_id')
        user_id = self._required_string(data, 'user_id')

        cursor.execute(
            '''
            UPDATE payments
            SET status = 'refunded'
            WHERE payment_id = ?
            ''',
            (payment_id,),
        )

        cursor.execute(
            '''
            UPDATE subscriptions
            SET status = 'cancelled',
                updated_at = CURRENT_TIMESTAMP
            WHERE user_id = ?
            ''',
            (user_id,),
        )

        cursor.execute(
            '''
            INSERT INTO outbox (
                event_key,
                event_type,
                payload,
                status,
                created_at
            )
            VALUES (?, 'subscription.cancelled', ?, 'pending',
                    CURRENT_TIMESTAMP)
            ON CONFLICT(event_key) DO NOTHING
            ''',
            (
                f'subscription-cancelled:{payment_id}',
                json.dumps(
                    {
                        'user_id': user_id,
                        'payment_id': payment_id,
                    }
                ),
            ),
        )

    def _mark_event_processed(
        self,
        cursor: sqlite3.Cursor,
        event_id: str,
    ) -> None:
        cursor.execute(
            '''
            UPDATE webhook_events
            SET status = 'processed',
                processed_at = CURRENT_TIMESTAMP
            WHERE event_id = ?
            ''',
            (event_id,),
        )

    def _mark_event_ignored(
        self,
        cursor: sqlite3.Cursor,
        event_id: str,
    ) -> None:
        cursor.execute(
            '''
            UPDATE webhook_events
            SET status = 'ignored'
            WHERE event_id = ?
            ''',
            (event_id,),
        )

    def _validate_event_time(
        self,
        event_created_at: int,
        received_at: datetime,
    ) -> None:
        received_timestamp = int(
            received_at.astimezone(timezone.utc).timestamp()
        )

        difference = abs(received_timestamp - event_created_at)

        if difference > self.max_clock_skew_seconds:
            raise InvalidPayloadError('Event timestamp is too old')

    @staticmethod
    def _required_string(
        payload: dict[str, Any],
        key: str,
    ) -> str:
        value = payload.get(key)

        if not isinstance(value, str) or not value.strip():
            raise InvalidPayloadError(f'Invalid field: {key}')

        return value

    @staticmethod
    def _required_integer(
        payload: dict[str, Any],
        key: str,
    ) -> int:
        value = payload.get(key)

        if isinstance(value, bool) or not isinstance(value, int):
            raise InvalidPayloadError(f'Invalid field: {key}')

        return value


def create_schema(connection: sqlite3.Connection) -> None:
    connection.executescript(
        '''
        CREATE TABLE IF NOT EXISTS webhook_events (
            event_id TEXT PRIMARY KEY,
            event_type TEXT NOT NULL,
            payload BLOB NOT NULL,
            status TEXT NOT NULL,
            created_at TEXT NOT NULL,
            processed_at TEXT
        );

        CREATE TABLE IF NOT EXISTS payments (
            payment_id TEXT PRIMARY KEY,
            user_id TEXT NOT NULL,
            plan TEXT NOT NULL,
            status TEXT NOT NULL,
            created_at TEXT NOT NULL
        );

        CREATE TABLE IF NOT EXISTS subscriptions (
            user_id TEXT PRIMARY KEY,
            plan TEXT NOT NULL,
            status TEXT NOT NULL,
            updated_at TEXT NOT NULL
        );

        CREATE TABLE IF NOT EXISTS outbox (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            event_key TEXT NOT NULL UNIQUE,
            event_type TEXT NOT NULL,
            payload TEXT NOT NULL,
            status TEXT NOT NULL,
            created_at TEXT NOT NULL
        );
        '''
    )


def build_signature(secret: bytes, body: bytes) -> str:
    return hmac.new(secret, body, hashlib.sha256).hexdigest()


if __name__ == '__main__':
    database = sqlite3.connect('webhooks.db')
    create_schema(database)

    secret = b'local-development-secret'
    now = int(time.time())

    body = json.dumps(
        {
            'id': 'evt_2026_08_05_001',
            'type': 'payment.completed',
            'created_at': now,
            'data': {
                'payment_id': 'pay_9001',
                'user_id': 'user_42',
                'plan': 'pro',
            },
        }
    ).encode('utf-8')

    processor = WebhookProcessor(database, secret)

    result = processor.process(
        WebhookRequest(
            raw_body=body,
            signature=build_signature(secret, body),
            received_at=datetime.now(timezone.utc),
        )
    )

    duplicate_result = processor.process(
        WebhookRequest(
            raw_body=body,
            signature=build_signature(secret, body),
            received_at=datetime.now(timezone.utc),
        )
    )

    print(result)
    print(duplicate_result)

This version is longer and less elegant than the first one. That is normal. Correctness often looks slightly boring.

The event ID has a unique constraint. Claiming an event and changing the business state happen inside one transaction. External work is not performed directly inside the webhook request. Instead, an outbox record is written and can later be processed by a worker.

Is the implementation perfect? No. SQLite has different concurrency behaviour from a production PostgreSQL cluster. A real system also needs retry rules for outbox messages, observability, retention policies and a clear answer for events stuck in the processing state.

But now those questions are visible. The first generated version hid them behind a clean handler method.

Security bugs often begin as convenience

Another common pattern appeared when asking AI to generate small utilities around files, URLs and shell commands.

The generated code often optimises for the example input. If the prompt contains a normal file name, the model writes code for normal file names. If the prompt contains a trusted URL, it writes code for trusted URLs.

Attackers do not send example input.

A generated archive extractor may join the destination directory with an archive entry name without checking whether the result escapes the destination. A generated image downloader may accept arbitrary URLs and accidentally become a server-side request forgery tool. A generated command wrapper may pass user input into a shell because the code is shorter.

The strange part is that the same model can explain all three vulnerabilities perfectly when asked directly. It simply does not always apply that knowledge while generating an unrelated feature.

This looks less like forgetting and more like local optimisation. The model tries to complete the visible task. Security requirements that were not present in the immediate context become optional.

That is why adding a final prompt such as review this code for security problems helps but does not solve the deeper issue. The reviewer is still working from the same text and may preserve the same assumptions.

A better review starts with boundaries:

Which values come from users?
Which values come from another service?
Which values become file paths, queries, commands or URLs?
What can be repeated?
What can arrive out of order?
What happens after a timeout?
What state has already changed when an exception occurs?

These questions are less exciting than asking for a complete application in one prompt. They are also much cheaper than an incident.

Cancellation and timeouts create invisible half-finished operations

Async code is another place where generated solutions look better than they behave.

Imagine a request handler that starts three tasks: save a file, update metadata and notify another service. The AI uses Promise.all in JavaScript or asyncio.gather in Python. Nice and compact.

Then one task fails.

What happens to the others? Have they already changed state? Can they be cancelled? Is cancellation safe? Will retrying the complete operation duplicate anything?

AI-generated async code often treats concurrency as a performance feature. In production, concurrency is also a failure model.

A timeout does not mean the remote operation did not happen. It only means the caller stopped waiting. This distinction is easy to say and easy to forget while reading clean code.

The same problem appears with database calls. A client may time out while the database transaction is still committing. Retrying blindly can repeat an operation whose result is merely unknown.

This is why mature systems sometimes return an operation ID before the final result exists. The caller can query the status instead of guessing whether a timeout means failure. It is also why idempotency keys matter far beyond payment APIs.

When reviewing generated async code, try replacing every await with a small question:

What if this completed, but the next line never ran?

That one question found more bugs in my experiments than another round of code generation.

The most dangerous function may be the test helper

One failure took longer to notice because the production code was mostly correct. The bug lived in the generated test fixture.

The fixture created a database, inserted records and returned a repository. Tests modified that repository, but cleanup ran only when the test completed normally. A failed assertion left state behind. Later tests inherited it and produced random failures depending on execution order.

The AI then tried to fix the flaky tests by adding delays.

That was funny for about ten seconds.

Generated test utilities often deserve the same review as production infrastructure. Look for shared mutable state, fixed ports, real clocks, random values without saved seeds, cleanup outside finally blocks and mocks that return impossible combinations.

A mock can also make invalid behaviour look valid. For example, a repository mock may accept two writes that a real database would reject because of a unique constraint. A unit test then proves that the service works in a universe where its database has no rules.

Useful tests should attack assumptions, not repeat examples. For generated code, I now add at least a few tests from these groups:

the same command runs twice
two commands run concurrently
dependencies fail after partial progress
input contains empty, huge or malformed values
events arrive in the wrong order
time moves across a boundary
cleanup runs after cancellation
the database rejects a write

Not every small project needs all of them. But any project that changes money, permissions, files or external state probably needs more than a happy-path unit test.

What experienced programmers still provide

It is tempting to describe AI as a junior developer. That comparison is not quite right.

A junior developer usually knows when they are unsure. They ask why a transaction is needed. They remember the production bug from last week. They can learn that one strange business rule which exists because an old customer sends duplicate files every Friday.

The model has no scar tissue.

It has patterns, and many patterns are excellent. It can generate a parser, a migration, a test skeleton or a boring adapter faster than most people. But it does not naturally carry the history of your system. It does not know which apparently harmless shortcut caused an outage two years ago unless someone puts that knowledge into the context.

Experienced programmers do more than write advanced syntax. They recognise dangerous shapes.

A read followed by a write.

A check followed by an action.

A side effect before a commit.

A retry without an idempotency key.

A timeout treated as a confirmed failure.

A test double that behaves better than reality.

None of these patterns looks dramatic. That is exactly why they matter.

The useful skill is no longer producing every line manually. It is knowing which lines cannot be trusted merely because they compile.

My current rules for working with generated code

After breaking enough generated examples, I stopped asking AI for complete solutions without constraints. The workflow is now less magical but much more useful.

First, define invariants before implementation. A balance cannot become negative. One event changes state at most once. A user cannot access another user’s object. A failed operation leaves either the old state or the complete new state.

Second, ask for failure behaviour. What happens on a duplicate request, timeout, partial database failure, cancellation or malformed input?

Third, test at the boundaries between components. A method-level unit test will not reveal a race between two requests or a mismatch between application logic and a database constraint.

Fourth, review generated tests separately. Tests are code, not evidence from an independent witness.

Fifth, remove decorative complexity. AI likes wrappers, factories and interfaces because they are common in training examples. Every abstraction should earn its place by protecting a boundary or simplifying a real change.

Finally, run adversarial experiments. Send the same request fifty times. Cancel it halfway through. Reverse the event order. Make one dependency fail after another succeeds. Use a file name that tries to leave its directory. Use Unicode where ASCII was expected.

The goal is not to prove that AI writes bad code. Humans write every class of bug described here, often with fewer comments.

The goal is to stop confusing polished output with verified behaviour.

AI-generated code can be an excellent first draft. But first drafts should make us curious, not relaxed.

When the code looks finished after thirty seconds, that is usually the best moment to start asking what it forgot.

The AI Code Passed Every Test. Then Two Requests Arrived at the Same Time