Foundations
Establish model inputs, context, and engineering foundations so the rest of the system starts from explicit, verifiable contracts.
What information does the model receive, and how does the developer create reliable inputs and engineering foundations?Concept calibration
Context
- What it is
- The information visible to a model call, including instructions, conversation, tool schemas and results, retrieved evidence, and current state.
- What it is not
- It is not merely chat history, nor everything the model can remember permanently.
Observability
- What it is
- Use structured logs, traces, metrics, and version data to explain what happened during a system run.
- What it is not
- It is not just terminal output; logs without correlation IDs, state transitions, and privacy boundaries are hard to diagnose.
Python, Git, HTTP & Engineering Foundations
Build the Forge CLI, configuration, schemas, HTTP client, async flow, logging, and tests.
Python 3.12 · pytest · Pydantic · SQLite · deterministic MockConcept calibration\n\nPut the concepts used in this module back inside their engineering boundaries before you work with the code.\n\n\n\n### Context context\n\n- What it is: The information visible to one model call, including instructions, conversation excerpts, tool schemas and results, retrieved evidence, current state, and output constraints.\n- What it is not: It is not merely chat history or every item in a database; only information selected for the current call belongs to its context.\n- Relationship to adjacent concepts: RAG selects external evidence, memory preserves cross-step state, and tool schemas describe actions. All can supply context, but their responsibilities differ.\n- Where it lives in HeatStack Forge: Forge assembles context by task, permission, budget, and source priority while recording provenance and version for each item.\n- Typical misuse and correction: A common misuse equates more tokens with better understanding. Use selection, truncation, citation, and sensitive-data filters, then verify them on fixed fixtures.\n\n### Observability observability\n\n- What it is: Structured logs, traces, metrics, state transitions, and version information that explain a run and support failure diagnosis and version comparison.\n- What it is not: It is not merely terminal output or unlimited input recording; logs without correlation IDs, state transitions, or privacy boundaries are hard to diagnose and may leak data.\n- Relationship to adjacent concepts: The harness emits runtime events, the agent loop and tool calls form traces, evaluation aggregates metrics, and memory and recovery depend on versioned records.\n- Where it lives in HeatStack Forge: Forge creates an operation_id for each task and records stage, tool, latency, cost, error class, state change, and evidence path while filtering sensitive data.\n- Typical misuse and correction: A common misuse adds logs only after failure. Define events and correlation fields with the contract, then use fault injection to prove the timeline can be reconstructed.\n\n
Establish an importable, testable Forge layout that separates source, tests, and configuration
Start with a concrete problem
Why can’t a plain directory be imported as a Python package?
You have inherited a directory named forge that currently contains only an empty init.py file and no project_layout.py.
The test file tests/test_stage.py attempts to import inspect_layout and is_importable_package from forge.project_layout, but the missing project_layout.py causes an import error.
You need to create forge/project_layout.py and implement two functions: inspect_layout checks required project paths and returns a list of missing paths, and is_importable_package determines whether the forge directory is a real Python package.
The key decision is that determining whether a directory is a package cannot rely solely on the directory’s existence; you must check for the package marker file init.py.
Make a prediction
When implementing is_importable_package, which check correctly determines that the forge directory is an importable Python package?
- Check whether (root / “forge”).is_dir() is true
- Check whether (root / “forge” / “init.py”).is_file() is true
- Check whether (root / “forge”).exists() is true
- Check whether (root / “forge”).iterdir() is non-empty
Reasoning guide: The correct answer is to check whether the init.py file exists. A Python package must contain an init.py file; a directory alone is not sufficient for an import statement to succeed.
What it is
The essence of an importable package
A Python package is a directory containing an init.py file, which serves as the package’s initialization entry point and allows the directory to be recognized by import statements.
The inspect_layout function validates the project structure by checking a set of required paths and returns a list of missing paths, helping developers quickly locate issues.
What it is not
Common misconceptions
A plain directory, even if it has the same name as the package, cannot be imported as a package because the Python interpreter does not treat it as a module.
The list returned by inspect_layout is not an error message but a collection of missing paths; an empty list indicates that all required paths exist.
How it relates to neighboring concepts
Relationships between functions
is_importable_package focuses only on whether the forge directory is a package, while inspect_layout checks the entire project’s required paths, including the package marker file, test directory, and configuration file.
Both rely on path existence checks, but is_importable_package specifically validates the package marker file, whereas inspect_layout provides a broader project structure validation.
Boundary decision for this stage: When determining whether a directory is a Python package, you must check for the existence of the init.py file, not merely the existence of the directory itself.
Load environment configuration from an explicit allowlist and redact secrets from diagnostics
Start with a concrete problem
Public configuration leaks the API key
In forge/config.py, the public_config function directly uses dataclasses.asdict to return the internal configuration dictionary, causing the raw api_key value to appear in diagnostic output.
The test test_public_config_never_contains_api_key asserts that repr(shown) does not contain secret-value and that shown["api_key"] equals [redacted].
The current implementation fails this test because asdict copies all fields verbatim, including the sensitive field.
You need to modify public_config to return an explicitly constructed dictionary where api_key is replaced with a redaction marker.
Make a prediction
What happens if you directly use dataclasses.asdict(config) to generate the public configuration?
- The public configuration will contain the raw API key, causing a secret leak.
- The public configuration will automatically redact the API key without extra handling.
- The public configuration will raise an exception because
asdictdoes not support frozen dataclasses. - The public configuration will omit the
api_keyfield and return only the other fields.
Reasoning guide: The correct answer is the first option. asdict recursively converts the dataclass to a dictionary and preserves the original values of all fields, so api_key appears in plaintext in the public configuration.
What it is
What the public configuration is
The public configuration is a view of configuration intended for diagnostics, logs, or user interfaces, and it must hide sensitive information.
It is built by explicitly constructing a dictionary to precisely control which fields are exposed and how they are transformed.
In this stage, the public configuration contains api_base, model, and the redacted api_key.
What it is not
What the public configuration is not
It is not a complete copy of the internal configuration and cannot be generated directly with asdict.
It is not the configuration used for actual API calls; those still use the internal Config object.
It does not validate configuration completeness; validation is performed by load_config.
How it relates to neighboring concepts
Relationships with other components
load_config reads required variables from the environment dictionary and constructs the internal Config object.
public_config receives the internal Config object and returns a safe public dictionary.
Tests create a configuration via load_config and then call public_config to verify the redaction behavior.
Boundary decision for this stage: The public configuration must be explicitly constructed with sensitive fields replaced by placeholders; internal configuration and diagnostic output must be separated.
Parse external dictionaries into range-checked request contracts and reject unknown scenarios
Start with a concrete problem
External dictionaries may contain unknown scenarios or out-of-range step counts
The file forge/contracts_v2.py does not exist yet, so the tests test_unknown_scenario_is_rejected and test_step_budget_has_a_bounded_range in tests/test_stage.py fail with an import error.
You need to implement ForgeRequest.parse, which takes an external dictionary and returns a ForgeRequest instance, but must reject unknown scenarios (like finance) and out-of-range max_steps (like 100).
If you only perform type conversion, finance would be accepted as a valid scenario and max_steps=100 would also be accepted, leading to invalid configurations downstream.
Therefore, parsing must explicitly check that the scenario belongs to an allowed set and that max_steps is between 1 and 20.
Make a prediction
What happens if ForgeRequest.parse only uses str() and int() to convert inputs?
- Unknown scenarios are rejected because type conversion automatically validates.
- Unknown scenarios are accepted because type conversion does not check semantics.
max_stepsis automatically limited to between 1 and 20.- Parsing raises
ValueErrorbecause dictionary keys are missing.
Reasoning guide: The correct answer is the second option: type conversion only ensures the type is correct, it does not check whether the value belongs to an allowed set or range. Therefore, finance would be accepted and max_steps=100 would also be accepted.
What it is
A request contract is a parser with semantic validation
ForgeRequest.parse is a class method that takes an external dictionary, extracts and converts fields, then performs semantic validation: the scenario must be in ALLOWED_SCENARIOS, the request length must be at least 8 characters, and max_steps must be between 1 and 20.
Only after validation passes does it construct and return an immutable ForgeRequest instance, ensuring downstream code can safely use these fields.
What it is not
Type conversion is not validation
str() and int() only perform type conversion; they do not check whether the value is legal. For example, str('finance') returns 'finance', but 'finance' is not in the allowed set.
Similarly, int('100') returns 100, but 100 is outside the legal range for max_steps. Therefore, explicit checks must be written.
How it relates to neighboring concepts
Relationship between parsing, validation, and construction
The parsing process has three steps: first extract raw values from the dictionary and convert to expected types, then perform semantic validation, and finally construct the dataclass instance.
When validation fails, a ValueError is raised, preventing invalid data from entering the system; when validation passes, the constructed instance is immutable, ensuring consistency for later use.
Boundary decision for this stage: When the input dictionary is missing fields or has wrong types, should parse raise an exception or use defaults? This stage chooses to use defaults (like empty string or 0), but subsequent validation will reject these defaults, ensuring only valid requests pass.
Send JSON through a replaceable transport and explicitly handle status codes and response types
Start with a concrete problem
The HTTP client must reject non-success status codes and non-object response bodies
The file forge/http_client.py does not exist yet, so the tests test_non_success_status_is_not_treated_as_data and test_transport_receives_timeout_and_normalized_url fail during collection with ModuleNotFoundError.
You need to create ApiClient and Response from scratch, and make create_plan raise a RuntimeError containing ‘429’ when it receives a 429 status code, instead of returning the error response body as normal data.
Additionally, create_plan must strip the trailing slash from base_url, append /plans, and pass the timeout given at construction unchanged to transport.post.
Make a prediction
If create_plan directly returns response.body, what happens when the server returns a 429 rate-limit response?
- The caller receives an error dictionary but cannot distinguish it from successful data.
ApiClientautomatically raises an exception because the HTTP library handles status codes.- The program crashes because
response.bodyis not a dictionary. transport.postretries until it succeeds.
Reasoning guide: The correct option is the first one: directly returning response.body passes the error response body to the caller as if it were normal data, so the caller cannot tell from the return value whether the request succeeded. The second option is wrong because ApiClient uses a custom Transport protocol and does not automatically check status codes. The third option is wrong because response.body can be any object, not necessarily a dictionary. The fourth option is wrong because the Transport protocol does not define retry behavior.
What it is
Transport success and application validation are separate
The Transport protocol only sends the request over the network and returns a Response object; it does not interpret the meaning of the status code.
After receiving the Response, ApiClient must check that status is between 200 and 299 and that body is a dictionary before passing the data to upper business logic.
What it is not
Not every HTTP response can be returned directly as data
A non-2xx status code means the request did not succeed; the response body is usually an error message rather than business data, and returning it directly would mislead the caller into thinking the operation succeeded.
When the response body is not a dictionary, upper-level code cannot safely access fields, so it must be rejected early to avoid later AttributeError or type errors.
How it relates to neighboring concepts
Collaboration among Transport, Response, and ApiClient
ApiClient depends on the Transport protocol to send requests; Transport returns a Response containing status and body.
ApiClient.create_plan passes the business payload and timeout to transport.post, then checks the status code and type of the returned Response, and finally returns body.
Boundary decision for this stage: When response.status is not between 200 and 299, raise RuntimeError including the status code; when response.body is not a dictionary, raise ValueError. Only when both checks pass can response.body be returned.
Run independent reads concurrently while bounding in-flight work and preserving result order
Start with a concrete problem
Why does gather start all tasks at once?
In forge/concurrency.py, you are asked to implement bounded_map, which takes an iterable, an async worker function, and a concurrency limit limit, and returns a list of results in the same order as the input.
The starting code does not have this file, so the tests fail immediately when importing forge.concurrency with ModuleNotFoundError: No module named 'forge.concurrency'.
Even if you create the file and naively use asyncio.gather to start all workers, the test test_in_flight_work_never_exceeds_limit will fail because the peak concurrency reaches 6 instead of the expected 2.
You need to understand that asyncio.gather itself does not limit concurrency; you must explicitly use asyncio.Semaphore to constrain the number of simultaneously running tasks.
Make a prediction
If you directly use asyncio.gather(*(worker(item) for item in items)) without any restriction, when limit=2 and there are 6 input elements, what will the peak concurrency be?
- 2, because the limit parameter will automatically take effect
- 6, because gather will start all coroutines at once
- 1, because asyncio executes serially by default
- Uncertain, depends on event loop scheduling
Reasoning guide: The correct answer is 6. asyncio.gather immediately creates all coroutines and schedules them concurrently; it does not read or apply any limit parameter. Only by explicitly using asyncio.Semaphore can you limit the number of tasks running simultaneously.
What it is
What bounded concurrency is
Bounded concurrency is a control pattern: you have a set of independent tasks, but only at most N tasks are allowed to execute simultaneously; the rest must wait.
In asyncio, asyncio.Semaphore is a counter initialized to N; each task acquires a permit before starting via async with semaphore, and if no permits are available, it suspends until one is released.
bounded_map wraps each worker call in a coroutine protected by the semaphore, then uses asyncio.gather to collect all wrapped coroutines, thereby both preserving result order and limiting concurrency.
What it is not
What bounded concurrency is not
It is not a built-in feature of asyncio.gather: gather only handles concurrent scheduling and result aggregation, and provides no concurrency limit.
It is not simple serial execution: serial execution runs one task at a time, while bounded concurrency allows up to N tasks to run simultaneously, balancing resource utilization and system pressure.
It is not an ad-hoc solution using asyncio.sleep or manual counting; the correct approach is to use the standard library synchronization primitive asyncio.Semaphore.
How it relates to neighboring concepts
Relationship with other concepts
asyncio.Semaphore is similar to asyncio.Lock, but Lock allows only one holder, while Semaphore allows multiple holders, the number determined by its initial value.
The relationship between bounded_map and asyncio.gather is: gather concurrently executes all wrapped coroutines, while each wrapped coroutine internally controls when the actual worker starts via the semaphore.
Result order is guaranteed by gather: it collects return values in the order the coroutines were passed, so even if tasks complete in a different order, the final list order matches the input order.
Boundary decision for this stage: When you need to run multiple asynchronous tasks concurrently but system resources are limited, you must use asyncio.Semaphore to explicitly limit concurrency; if the number of tasks is small and resources are abundant, you can use asyncio.gather directly without a limit.
Emit classifiable structured events and recursively redact nested sensitive fields
Start with a concrete problem
Nested sensitive information leaks in log events
In the forge system, when a provider call fails, we need to record a structured event containing an operation ID, error class, and details.
The details may contain nested dictionaries and lists, such as an authorization field in request headers or a token field in list items.
If only top-level dictionary keys are redacted, nested sensitive values will still appear in logs, causing secret leakage.
The current test test_nested_secrets_are_redacted requires nested authorization and token values to be replaced with [redacted], but the starting code does not implement recursive redaction.
Make a prediction
When implementing the redact function, what happens to nested sensitive fields if only the top-level dictionary is processed?
- Nested sensitive fields are automatically redacted because the top-level dictionary has already been processed.
- Nested sensitive fields remain unchanged, causing a leak.
- Nested sensitive fields are deleted and will not appear in the result.
- The program raises an exception because it cannot handle nested structures.
Reasoning guide: The correct answer is the second option: nested sensitive fields remain unchanged. Because the redact function only iterates over the top-level dictionary’s key-value pairs, it does not recursively process values that are dictionaries or lists, so nested sensitive keys are not recognized and redacted.
What it is
What recursive redaction is
Recursive redaction is a method of traversing arbitrarily nested data structures (dictionaries and lists) and replacing sensitive key values at every level.
It ensures that no matter how deep sensitive information is located in the data structure, it is replaced with [redacted], preventing leakage.
The core of recursive redaction is that the function calls itself to process substructures until it reaches primitive types (such as strings, numbers).
What it is not
What recursive redaction is not
Recursive redaction is not merely checking keys of the top-level dictionary, nor is it a simple string replacement over the entire data structure.
It does not delete sensitive fields; instead, it preserves the key names and replaces the values with a placeholder to maintain the event structure’s integrity.
It is not a shallow check hardcoded for specific key names, but a generic process based on a set of sensitive keys.
How it relates to neighboring concepts
Relationship with other concepts
Recursive redaction is closely related to data structure traversal: dictionaries require iterating over key-value pairs, and lists require iterating over elements.
It works with the event generation function event, which calls redact to process the details field, ensuring the output event contains no sensitive information.
It is directly related to the test test_nested_secrets_are_redacted, which verifies that nested sensitive values are correctly redacted.
Boundary decision for this stage: The boundary of recursive redaction is: only process two container types, dictionaries and lists; for other types (such as strings, numbers, booleans), return the original value directly. Sensitive key detection is based on the SECRET_KEYS set, and key comparison is case-insensitive.
Combine commit identity, test results, and workspace state into an explainable release gate
Start with a concrete problem
Why can a release fail even when all tests pass?
You are implementing the release gate for the forge project. Currently, forge/release_gate.py does not exist, so the tests test_release_requires_commit_and_clean_generated_artifacts and test_passing_release_has_no_reasons fail with an import error.
The release gate must check commit identity, test results, and generated files simultaneously. If you only check whether tests pass, a release with no commit SHA and containing debug.log would be incorrectly accepted.
You need to implement the evaluate_release function so that it returns (passed, reasons), where passed is a boolean and reasons is a list of strings. When the commit SHA is shorter than 7 characters, there is a failed check, or there are generated files ending in .pyc or .log, passed must be False and specific reasons must be provided.
Make a prediction
Before implementing evaluate_release, predict: what happens if you only check whether all tests pass and ignore the commit SHA and generated files?
- The release gate will correctly reject all dirty releases.
- The release gate will incorrectly accept a release with no commit SHA and containing debug.log.
- The release gate will crash because it cannot handle an empty commit SHA.
- The release gate will ignore test results and only check generated files.
Reasoning guide: The correct answer is the second option. Checking only test pass status ignores commit identity and workspace hygiene, causing an untraceable dirty release to be accepted. This is exactly the defect injected in the failure fixture.
What it is
What a release gate is
A release gate is a function that aggregates multiple check results and decides whether to allow a release. It receives a commit SHA, a list of check results, and a list of changed files, and returns a boolean and a list of reasons.
It must verify commit identity, test results, and workspace hygiene simultaneously. Missing any of these blocks the release, even if all tests pass.
What it is not
What a release gate is not
A release gate is not a simple test-pass checker. It cannot just return all(check.passed) because that would ignore commit identity and generated files.
It is also not a logging system or a build tool. It only evaluates release conditions and does not execute tests or clean files.
How it relates to neighboring concepts
Relationships with other components
The release gate uses the CheckResult dataclass to represent each check’s result. CheckResult contains name, passed, and detail fields.
The release gate is directly related to the test file tests/test_stage.py. The test file defines the contract the release gate must satisfy, including checks for commit SHA and generated files.
Boundary decision for this stage: The release gate must check commit identity, test results, and generated files; missing any of these blocks the release.
Complete the chapter
Ship a runnable forge CLI, configuration, logs, and test baseline.
Local lab self-check
- Not started
- 2Reading
- 3Lab downloaded
- 4Test result read
- 5Local check passed
verification.json is parsed only in this browser and is never uploaded. This checks the lab contract; it is not server certification or third-party endorsement.
Concept calibration and one-week review
Three questions check the chapter's boundaries, delivery evidence, and recovery path. Answers stay in this browser.