Set task acceptance criteria and request resource budgets

Require both a correct task result and a stated request-resource policy. Connect the existing application checks to byte limits and capacity admission, then inspect which limits prevent execution and which withhold its output.

Problem

When you need this. You need explicit success criteria and request limits before comparing implementations.

A response uses 147 bytes but gives the staging command for a production request. It fits a resource budget and fails the application task. Conversely, the correct 167-byte result can violate a policy that permits only one output byte. Neither dimension can replace the other.

You will combine the existing application checks with source-byte, output-byte and capacity limits. Seven local scenarios exercise successful handling, wrong results and refusal before invocation. The numbers below come from the Python function and local gate; there is no model call or latency benchmark.

Concepts

Task acceptance asks whether the returned result satisfies the user’s contract. Resource compliance asks whether the wrapper stayed within a declared limit. Both must pass; a small wrong answer is still wrong.

Admission grants a lease before application work starts. A queue-wait limit controls waiting for that lease. Source bytes can be checked before application invocation; serialized output bytes are known afterwards. None of these limits is a provider-token allowance, a price ceiling or an execution deadline.

Architecture

The wrapper reads and checks source bytes, acquires a lease from one shared capacity gate, and invokes the existing application. Graders inspect the result, then the wrapper serializes it and checks output size. Preflight refusal leaves the function uncalled; an oversized post-execution result is withheld without altering the command.

Working code

Declare what each limit actually covers

The fixture policy is small enough to inspect:

{
  "max_source_bytes": 1024,
  "max_output_bytes": 512,
  "max_active": 1,
  "max_waiting": 1,
  "wait_seconds": 0.02
}

max_source_bytes covers the exact source snapshot read for the lookup, including whitespace. max_output_bytes covers the canonical UTF-8 JSON produced by this wrapper. Neither counts model tokens, HTTP headers or all process memory. Source reading happens before the byte check, so the cap prevents application invocation, not the allocation needed to read the file.

The active and waiting counts belong to one shared capacity gate. wait_seconds limits time spent waiting for a lease. It is not an execution deadline. A provider timeout, cancellation policy or monetary limit would need separate implementation at that dependency boundary.

These are authored policy values, not measurements of what a deployed service can sustain. A service needs workload and capacity measurements before choosing its own values.

Download the complete lab and enter ai-system-acceptance/:

python3.12 -m venv .venv
source .venv/bin/activate
python inspect_acceptance.py
python -m unittest -v test_acceptance.py

The ZIP includes the existing architecture, cases and graders, plus the unchanged RAG capacity gate. It uses the standard library, local threads and an asyncio event loop. No package, model or data download is required.

Refuse work before invoking the application when possible

The wrapper checks source size before parsing and invocation. It then acquires a capacity lease before calling the function. An oversized source, full queue or expired queue wait leaves called: false and the task verdict not_run.

"""Join existing task checks and capacity admission at an application boundary."""
from dataclasses import asdict, dataclass
import json
from pathlib import Path
import sys
sys.path.insert(0, str(Path(__file__).resolve().parent / 'vendor'))
from gate import Gate, Refused as CapacityRefused, finish_worker
import harness
from architecture import Request, answer
import task


@dataclass(frozen=True)
class Policy:
    max_source_bytes: int = 1024
    max_output_bytes: int = 512
    max_active: int = 1
    max_waiting: int = 1
    wait_seconds: float = 0.02

    def __post_init__(self):
        for field, maximum in [('max_source_bytes', 1_048_576), ('max_output_bytes', 65_536)]:
            if type(getattr(self, field)) is not int or not 1 <= getattr(self, field) <= maximum:
                raise ValueError('invalid byte budget: ' + field)
        Gate(self.max_active, self.max_waiting, self.wait_seconds)

    def gate(self):
        return Gate(self.max_active, self.max_waiting, self.wait_seconds)


def encode(value):
    return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(',', ':'), allow_nan=False).encode('utf-8')


def invoke(case, pages):
    return answer(Request(**case['input']['request']), pages)


async def handle(case, raw_source, policy, gate, application=invoke):
    if not isinstance(policy, Policy) or not isinstance(gate, Gate) or any(getattr(gate, field) != getattr(policy, field) for field in ('max_active', 'max_waiting', 'wait_seconds')):
        raise ValueError('capacity gate must match the declared policy')
    if not isinstance(raw_source, bytes):
        raise ValueError('source snapshot must be bytes')
    result = {'called': False, 'source_bytes': len(raw_source), 'output_bytes': 0,
              'quality': {'verdict': 'not_run'}, 'resource_status': 'not_admitted',
              'response': None, 'meets_contract': False, 'policy': asdict(policy)}
    if len(raw_source) > policy.max_source_bytes:
        return dict(result, resource_status='source_too_large')
    try:
        pages = json.loads(raw_source)
        if not isinstance(pages, list):
            raise ValueError('expected a source list')
    except (ValueError, UnicodeError):
        return dict(result, resource_status='source_invalid')

    def call_application():
        result['called'] = True
        return application(case, pages)

    try:
        async with gate.lease():
            try:
                output = await finish_worker(call_application)
            except Exception as error:
                quality = harness.judge_error(case, error)
                response = {'refused': str(error)} if quality['verdict'] == 'pass' else {'error': type(error).__name__}
            else:
                quality = ({'verdict': 'incomplete', 'reason': 'no output'} if output is None
                           else harness.grade(case, output, tuple(task.GRADERS) + harness.CORE_GRADERS))
                response = output
            try:
                produced = encode(response)
            except (TypeError, ValueError):
                return dict(result, quality=quality, resource_status='output_not_serializable')
            resource_status = 'within_limits' if len(produced) <= policy.max_output_bytes else 'output_too_large'
            meets = resource_status == 'within_limits' and quality['verdict'] == 'pass'
            return dict(result, output_bytes=len(produced), quality=quality, resource_status=resource_status,
                        response=response if meets else None, meets_contract=meets)
    except CapacityRefused as error:
        return dict(result, resource_status=str(error))

Create one gate for the shared worker capacity and pass it to requests using that capacity. Creating a fresh gate for every request would give each request its own allowance. The demonstration resets the gate between independent scenarios, then deliberately holds a lease within the saturation probes.

The wrapper also checks that the supplied gate matches the policy it reports. Otherwise a configured one-worker limit could appear in a record while a different gate admitted more work.

The reused finish_worker() runs this trusted local function in a thread and retains the lease while it finishes. Canceling the awaiting coroutine cannot safely be taken to mean that the underlying thread stopped. The existing gate lesson owns that lifecycle; this wrapper does not add a way to terminate arbitrary code.

Keep task checks separate from output accounting

After invocation, the existing graders check the response contract, source, exact command and action policy. A required refusal can be the correct task outcome. meets_contract means correct request handling under the declared limits; it never authorizes executing a rollback.

The task verdict is an evaluation verdict for a case with a known expected result. A production handler can reuse permission and schema checks, but it cannot assume an exact-answer oracle for an arbitrary user request.

The wrapper encodes the result and counts bytes after the function returns. If the result exceeds the output cap, it withholds the response. It does not cut a command in half to fit. It also withholds a response that fails the task checks, even when its byte count is small.

This output check cannot undo work already performed or money already spent. A real model adapter would need its own generation limit before the provider call and accounting for the provider’s actual usage. This lab makes neither a provider request nor a billing guarantee.

Malformed outputs keep their failed task verdict. If a value cannot be encoded as JSON, it is withheld with output_not_serializable rather than causing the report to disappear. The corresponding regression supplies a set as the function result.

Inspect the two independent verdicts

Declared request policy; actual local calls, byte counts and capacity outcomes
normal: called=True; quality=pass; resource=within_limits; bytes=167; meets_contract=True
wrong_command: called=True; quality=fail; resource=within_limits; bytes=147; meets_contract=False
oversized_source: called=False; quality=not_run; resource=source_too_large; bytes=0; meets_contract=False
output_limit: called=True; quality=pass; resource=output_too_large; bytes=167; meets_contract=False
queue_full: called=False; quality=not_run; resource=queue_full; bytes=0; meets_contract=False
wait_timeout: called=False; quality=not_run; resource=wait_timeout; bytes=0; meets_contract=False
required_refusal: called=True; quality=pass; resource=within_limits; bytes=61; meets_contract=True
Queue wait is bounded; execution time, memory and provider charges are not bounded by this policy.

The wrong-command probe passes resource checks and fails task checks. The output-limit probe does the reverse. The source and capacity probes do not invoke the function, so there is no task result to grade.

The queue probes use actual local admission. One holds the only active lease and allows no waiting slot, producing queue_full. The other permits one waiter whose declared wait expires, producing wait_timeout. Both finish with no active or waiting work. These outcomes test admission behavior; they do not establish a throughput or response-time target.

The source and output endpoint tests allow an exact byte-limit match and refuse one byte over. A separate test uses a non-ASCII value to distinguish encoded bytes from character count. Reporting the unit precisely matters more than choosing a familiar round number.

Decide what the application may promise

python inspect_acceptance.py --check
python inspect_acceptance.py --save my-policy-result.json

The result record keeps the policy, actual byte counts, invocation flag, task verdict and gate events for each scenario. It also binds the application, graders and reused gate to source hashes.

"""Exercise declared request limits with local calls and a real local queue wait."""
import argparse
import asyncio
from dataclasses import replace
import hashlib
import json
from pathlib import Path
from acceptance import Policy, handle, invoke, harness
import task

ROOT = Path(__file__).resolve().parent


async def experiment():
    policy = Policy(**json.loads((ROOT / 'policy.json').read_text()))
    cases = harness.load_cases(ROOT / 'cases.json', harness.load_task('task'))
    task.validate_inputs(cases)
    by_id = {case['id']: case for case in cases['cases']}
    case = by_id['v3-production-lookup']
    source = (ROOT / 'pages.json').read_bytes()
    results = {}

    async def run(name, chosen=policy, raw=source, selected=case, application=invoke, saturated=False):
        gate = chosen.gate()
        if saturated:
            async with gate.lease():
                result = await handle(selected, raw, chosen, gate, application)
        else:
            result = await handle(selected, raw, chosen, gate, application)
        result['capacity_after'] = gate.snapshot()
        results[name] = result

    await run('normal')

    def wrong_command(case, pages):
        output = invoke(case, pages)
        output['command_template'] = 'docuctl rollback RELEASE --env staging'
        return output

    await run('wrong_command', application=wrong_command)
    await run('oversized_source', raw=source + b' ' * policy.max_source_bytes)
    await run('output_limit', chosen=replace(policy, max_output_bytes=1))
    await run('queue_full', chosen=replace(policy, max_waiting=0), saturated=True)
    await run('wait_timeout', saturated=True)
    await run('required_refusal', selected=by_id['reader-proposal'])
    return {'mode': 'Local function calls and local capacity admission; no model, tokenizer or paid API',
            'assumption': 'Byte, capacity and queue-wait limits are declared scenario policy, not measured latency or monetary caps.',
            'files': {name: hashlib.sha256((ROOT / name).read_bytes()).hexdigest() for name in
                      ('architecture.py', 'pages.json', 'cases.json', 'task.py', 'acceptance.py', 'inspect_acceptance.py', 'policy.json', 'vendor/harness.py', 'vendor/gate.py')},
            'scenarios': results}


def record():
    return asyncio.run(experiment())


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('--save', type=Path)
    parser.add_argument('--check', action='store_true')
    args = parser.parse_args()
    result = record()
    if args.check and result != json.loads((ROOT / 'result.json').read_text()):
        raise SystemExit('acceptance results or source bindings differ')
    if args.save:
        args.save.write_text(json.dumps(result, indent=2) + '\n')
    print('Declared request policy; actual local calls, byte counts and capacity outcomes')
    for name, row in result['scenarios'].items():
        print(f"{name}: called={row['called']}; quality={row['quality']['verdict']}; "
              f"resource={row['resource_status']}; bytes={row['output_bytes']}; meets_contract={row['meets_contract']}")
    print('Queue wait is bounded; execution time, memory and provider charges are not bounded by this policy.')


if __name__ == '__main__':
    main()

This local boundary can promise source admission, bounded active/waiting counts, queue-wait handling and withheld invalid or oversized results. It cannot promise an execution deadline, total heap bound, provider-charge cap or correctness outside the supplied task checks. Those are separate requirements to implement and measure.

Design exercise: a product asks for “a two-second response.” Break that requirement into queue wait, application work, dependency calls and response delivery. Mark which parts this lab controls and where cancellation or an in-flight dependency could keep using capacity after the client leaves.

Failure cases

The wrong-command scenario fits the byte allowance and still fails acceptance. The oversized-output scenario produces the right command but withholds it. Saturation and expired queue waiting stop before invocation, while canceling an await does not prove that an already running worker thread has stopped.

Deployment

A service integration must create one gate per shared worker capacity and pass it to the requests using those workers. Record the effective limits with each outcome, and release leases according to actual worker completion. Add dependency timeouts and client-disconnection handling where those dependencies are invoked.

Scaling

Measure source reading, queue waiting, application work and response delivery separately. Increasing a queue limit can increase waiting without increasing throughput. The byte checks describe retained source and serialized output; measure actual process memory and downstream capacity before choosing service limits.

Key takeaways

• Task correctness and resource compliance are independent verdicts.
• Place a limit before the work it should prevent whenever the needed information is available.
• A queue-wait deadline does not terminate application execution.

Best practices

• Reuse one capacity gate for the resources it represents.
• Withhold oversized command output instead of truncating it.
• Record whether application work ran when a request is refused.

References

• Python, asyncio tasks: waiting, cancellation and running blocking functions in threads.
• Python, JSON encoding: JSON serialization and encoding choices.

Sources checked October 24, 2026. The byte counts and admission outcomes are from the bundled local execution; the policy values are scenario inputs.