All Learning Pathways

import anthropic

client = anthropic.Anthropic()

response = client.messages.create(
    model="claude-opus-4-1",
    max_tokens=1024,
    system="You are a concise weather assistant. Use the tool when asked about conditions.",
    tools=[
        {
            "name": "get_weather",
            "description": "Get the current weather for a given city.",
            "input_schema": {
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "City name, e.g. 'Nairobi'"}
                },
                "required": ["city"],
            },
        }
    ],
    messages=[{"role": "user", "content": "What's the weather in Nairobi?"}],
)

print(response.stop_reason)   # "tool_use" if Claude wants to call get_weather
print(response.content)       # list of content blocks, e.g. a tool_use block

# CLAUDE.md
## Project: payments-service

- Run tests with `pnpm test`, not `npm test`
- Never edit files under `generated/`, they're codegen output
- All new endpoints need a corresponding test in `tests/api/`
- Use the existing `Result<T, E>` type for error handling,
  not thrown exceptions
- Ask before adding a new dependency

---
name: pdf-form-filler
description: >
  Use when the user asks to fill out, complete, or populate a PDF form
  (tax forms, applications, intake forms) with provided data. Not for
  reading or summarizing PDF content, or for generating a new PDF from
  scratch.
---

# Filling PDF Forms

1. Identify the form fields with `scripts/list_fields.py <path>`.
2. Map the user's data onto field names; ask before guessing on
   ambiguous fields (e.g. multiple "Name" fields).
3. Fill with `scripts/fill_form.py <path> <field=value>...`.
4. Report which fields were left blank and why.

# Conceptual sketch of an MCP tool definition.
# The model never sees your code, only this description and schema.
# If the description is ambiguous, the model will call it wrong.

TOOL = {
    "name": "get_invoice",
    "description": (
        "Fetch a single invoice by its ID. Returns the invoice status, "
        "amount due, and due date. Use this before answering any question "
        "about whether a specific invoice has been paid."
    ),
    "input_schema": {
        "type": "object",
        "properties": {
            "invoice_id": {
                "type": "string",
                "description": "The invoice ID, e.g. 'INV-2024-0113'."
            }
        },
        "required": ["invoice_id"],
    },
}

def handle_call(name, arguments):
    if name == "get_invoice":
        return lookup_invoice(arguments["invoice_id"])
    raise ValueError(f"Unknown tool: {name}")

{
  "messages": [
    { "role": "system", "content": "You are a support agent for Acme Cloud. Always answer in three short bullet points." },
    { "role": "user", "content": "My deploy failed with error E-402, what do I do?" },
    { "role": "assistant", "content": "- E-402 means your build exceeded the memory limit\n- Increase the memory allocation in acme.yaml under 'build.resources'\n- Redeploy, and check the build logs if it fails again" }
  ]
}

# ReAct-style agent loop (pseudocode)
state = get_initial_observation()
for step in range(MAX_STEPS):
    thought = model.reason(goal, state, history)
    if thought.is_final_answer:
        return thought.answer

    action = thought.next_action          # e.g. call_tool("search", query)
    observation = execute(action)          # run it, catch failures
    history.append((thought, action, observation))

    if observation.failed and repeated_failure(history):
        break                              # bail out instead of looping forever
    state = update_state(state, observation)

return "stopped: exceeded max steps or unrecoverable failure"

{
  "name": "get_order_status",
  "description": "Look up the current shipping status of a customer order by its order ID. Use this when the user asks where their order is or whether it has shipped.",
  "parameters": {
    "type": "object",
    "properties": {
      "order_id": {
        "type": "string",
        "description": "The order ID, formatted like 'ORD-48213'. Do not guess this value, ask the user for it if unknown."
      }
    },
    "required": ["order_id"]
  }
}

def supervisor(task):
    plan = supervisor_llm.plan(task)
    results = {}

    for subtask in plan.subtasks:
        worker = select_worker(subtask.role)
        message = {
            "requested": subtask.instruction,
            "context": results,
        }
        results[subtask.role] = worker.run(message)

    return supervisor_llm.combine(results)

function step(agent_state, new_observation):
    context = agent_state.working_context + new_observation

    if token_count(context) > CONTEXT_BUDGET:
        old_part, recent_part = split(context, keep_recent=RECENT_STEPS)
        summary = llm_summarize(old_part)   # compress, don't discard
        context = summary + recent_part

    relevant_memories = retrieve(vector_store, query=new_observation, k=5)
    full_context = relevant_memories + context

    response = llm_call(full_context)

    if response.contains_durable_fact():
        write_to_long_term_memory(response.fact)

    return response, context

while not task_complete:
    observation = get_current_state()
    thought, action = model.decide(observation, goal, history)
    result = execute_tool(action, sandbox=True)
    history.append((thought, action, result))
    task_complete = check_goal_reached(result, goal)

class CostGuard:
    def __init__(self, max_calls_per_task=15, max_cost_per_user_per_day=2.00):
        self.max_calls_per_task = max_calls_per_task
        self.max_cost_per_user_per_day = max_cost_per_user_per_day
        self.spend_by_user = {}

    def check(self, user_id, task_call_count, projected_cost):
        if task_call_count >= self.max_calls_per_task:
            raise AgentHalted("task exceeded max tool calls, likely looping")

        spent_today = self.spend_by_user.get(user_id, 0.0)
        if spent_today + projected_cost > self.max_cost_per_user_per_day:
            raise AgentHalted("user daily cost cap reached, route to human")

        return True

    def record(self, user_id, actual_cost):
        self.spend_by_user[user_id] = self.spend_by_user.get(user_id, 0.0) + actual_cost


# every tool call goes through the guard before it fires
guard.check(user_id, task_call_count, projected_cost)
result = call_tool(...)
guard.record(user_id, result.cost)

# The core mechanical pattern behind every tool-using agent
  while iterations < MAX_ITERATIONS:
      response = model.generate(conversation_history)
  
      if response.type == "final_answer":
          return response.text
  
      # response.type == "tool_call"
      tool_name = response.tool_call.name
      arguments = response.tool_call.arguments  # raw, model-produced JSON
  
      if not is_valid(tool_name, arguments):       # your code, not the model
          result = {"error": "invalid arguments"}
      else:
          result = execute_tool(tool_name, arguments)  # your code actually runs it
  
      conversation_history.append(response)
      conversation_history.append({"role": "tool", "content": result})
      iterations += 1
  
  raise TimeoutError("Agent exceeded max iterations without a final answer")

// A persisted "task status" object: the plan's state, saved outside
  // any single conversation, so a fresh session can pick up mid-task.
  
  task_status = {
    goal: "Launch the Q3 customer research report",
    constraints: [
      "Budget capped at 40 interview hours",
      "Must use only publicly available survey tools"
    ],
    steps: [
      { id: 1, description: "Define research questions", status: "done" },
      { id: 2, description: "Recruit interview participants", status: "done" },
      { id: 3, description: "Conduct interviews", status: "in_progress" },
      { id: 4, description: "Synthesize findings", status: "pending" },
      { id: 5, description: "Write final report", status: "pending" }
    ],
    next_step_id: 3,
    decisions_log: [
      "Chose phone interviews over in-person to save travel time"
    ],
    last_updated: "2026-08-10T09:00:00Z"
  };
  
  // --- session ends here, maybe for days ---
  
  // A brand new session, possibly a different agent instance,
  // reloads the same object instead of starting from nothing:
  function resumeTask(saved_status) {
    const current = saved_status.steps.find(
      s => s.id === saved_status.next_step_id
    );
    return `Resuming "${saved_status.goal}". Current step: ${current.description}`;
  }

from sklearn.tree import DecisionTreeClassifier
from sklearn.model_selection import train_test_split

# X = input features, y = known labels (spam or not spam)
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42
)

model = DecisionTreeClassifier(max_depth=4)
model.fit(X_train, y_train)          # learn from labeled examples

predictions = model.predict(X_test)  # guess on unseen examples
accuracy = (predictions == y_test).mean()
print(f"Accuracy on held-out data: {accuracy:.2f}")

pick k (number of clusters)
randomly place k cluster centers

repeat until assignments stop changing:
    for each data point:
        assign it to the nearest cluster center
    for each cluster:
        recalculate its center as the average
        of all points currently assigned to it

import pandas as pd
from sklearn.preprocessing import OneHotEncoder, StandardScaler

df = pd.DataFrame({
    "color": ["red", "green", "blue", "green"],
    "income": [42000, 88000, 51000, 120000],
})

# Encoding: turn a categorical column into numeric columns a model can use
encoder = OneHotEncoder(sparse_output=False)
color_encoded = encoder.fit_transform(df[["color"]])

# Scaling: put a wide-range numeric column onto a comparable scale
scaler = StandardScaler()
income_scaled = scaler.fit_transform(df[["income"]])

# Without scaling, income (tens of thousands) would dwarf a 0/1 encoded
# feature in any distance-based or gradient-based model, not because it's
# more predictive, just because its raw numbers are bigger.
print(color_encoded, income_scaled)

# Computing precision and recall from confusion matrix counts
tp = 42   # true positives: correctly flagged as positive
fp = 8    # false positives: flagged positive, actually negative
fn = 5    # false negatives: missed, actually positive
tn = 945  # true negatives: correctly flagged as negative

precision = tp / (tp + fp)
recall = tp / (tp + fn)
accuracy = (tp + tn) / (tp + fp + fn + tn)

print(f"precision: {precision:.2f}")  # of flagged positives, how many were right
print(f"recall:    {recall:.2f}")     # of actual positives, how many were caught
print(f"accuracy:  {accuracy:.2f}")   # can look great even when recall is terrible

# simplified drift check: compare live prediction distribution
# against the distribution the model was trained on

def check_drift(training_dist, live_predictions, threshold=0.1):
    live_dist = bucket_counts(live_predictions)
    score = population_stability_index(training_dist, live_dist)

    if score > threshold:
        alert_team(
            message=f"Drift detected: PSI={score:.3f} exceeds {threshold}",
            severity="warning",
        )
        return False
    return True

# Chronological split: never shuffle time series data
  cutoff = int(len(series) * 0.8)
  train, test = series[:cutoff], series[cutoff:]
  
  # Expanding-window backtest: retrain as the window grows forward
  for cutoff in range(min_train_size, len(series), step):
      train, test = series[:cutoff], series[cutoff:cutoff + horizon]
      model.fit(train)
      preds = model.predict(horizon)
      score(preds, test)

# Toy matrix factorization: learn user and item embeddings
  # so that dot(user_vec, item_vec) approximates a rating.
  import numpy as np
  
  n_users, n_items, k = 500, 2000, 32
  user_emb = np.random.normal(scale=0.1, size=(n_users, k))
  item_emb = np.random.normal(scale=0.1, size=(n_items, k))
  
  def predict(u, i):
      return user_emb[u] @ item_emb[i]
  
  def train_step(u, i, rating, lr=0.01, reg=0.02):
      pred = predict(u, i)
      error = rating - pred
      # gradient step, nudging both vectors toward each other
      # when the interaction is positive
      user_emb[u] += lr * (error * item_emb[i] - reg * user_emb[u])
      item_emb[i] += lr * (error * user_emb[u] - reg * item_emb[i])

# Simplified non-maximum suppression (NMS)
# Input: a list of candidate boxes for ONE class, each with a confidence score

boxes = sort_by_confidence_descending(candidate_boxes)
kept = []

while boxes is not empty:
    best = boxes.pop(0)          # highest remaining confidence
    kept.append(best)

    # Remove any remaining box that overlaps "best" too much
    boxes = [b for b in boxes if overlap(b, best) < OVERLAP_THRESHOLD]

# "kept" now has roughly one box per real object, not thousands

// Object detection output: a handful of numbers per object
{ class: "person", box: [x_min: 120, y_min: 40, x_max: 260, y_max: 400], confidence: 0.94 }

// Segmentation output: a full grid matching the image dimensions
// every single pixel position gets its own predicted class
image size: 640 x 480
mask shape: [640, 480]   // one class label per pixel, not per object

mask[0..119][*]     = "background"
mask[120..260][40]  = "person"   // only the actual person-shaped pixels
mask[120..260][400] = "person"   // background between the legs stays "background"
mask[261..640][*]   = "background"

# Simplified: turning an image into a sequence of patch tokens

image = load_image()              # shape: (224, 224, 3)
patch_size = 16

patches = split_into_patches(image, patch_size)
# patches shape: (14 x 14 grid, 16, 16, 3) -> 196 total patches

tokens = [flatten(p) for p in patches]
# each token is a flat vector, like a "word" in a sentence

tokens = [linear_projection(t) for t in tokens]
tokens = add_position_embeddings(tokens)
# position embeddings matter here, unlike a CNN, the transformer
# has no built-in sense of where each patch sits in the image

output = transformer_encoder(tokens)   # standard self-attention stack
prediction = classification_head(output[0])  # the [CLS] token

# Converting stereo disparity into real-world depth

baseline        = distance between the two camera centers (meters)
focal_length    = camera focal length (pixels)
disparity       = how many pixels a point shifts between the left and right image

depth = (baseline * focal_length) / disparity

# Intuition:
# - larger disparity (big pixel shift) -> object is close -> smaller depth
# - smaller disparity (tiny pixel shift) -> object is far -> larger depth
# - disparity near 0 -> depth approaches infinity (point is effectively at the horizon)

"I don't like tokenization" tokenized into subwords:

["I", "don", "'t", "like", "token", "ization"]

Notice: "don't" splits into "don" + "'t", and
the rarer word "tokenization" splits into two
common pieces instead of getting its own slot.

Raw audio pipeline (conceptual):

1. Microphone captures a continuous sound wave (air pressure over time)
2. Sampling: the wave is measured thousands of times per second
   e.g. 16,000 samples/sec -> a long list of numbers (amplitudes)
3. Spectrogram conversion: chop audio into short time windows,
   measure which frequencies are present in each window
   -> produces a 2D grid: time (x-axis) by frequency (y-axis)
4. The spectrogram is now image-like data a neural network can process
5. ASR model maps spectrogram patterns -> phonemes -> words -> text
6. TTS runs this in reverse: text -> predicted spectrogram -> waveform

{
  "intent": "book_flight",
  "slots": {
    "origin": "Chicago",
    "destination": null,
    "date": "2026-09-12",
    "passengers": 1
  },
  "missing_slots": ["destination"],
  "next_action": "ask_for_destination",
  "turn_count": 3
}

# Simplified CLIP-style contrastive training loop
for images, captions in dataloader:
    image_embeds = image_encoder(images)      # shape: [batch, dim]
    text_embeds = text_encoder(captions)       # shape: [batch, dim]

    image_embeds = normalize(image_embeds)
    text_embeds = normalize(text_embeds)

    # similarity between every image and every caption in the batch
    similarity = image_embeds @ text_embeds.T  # [batch, batch]

    # the correct pairs sit on the diagonal, everything else is a mismatch
    labels = range(batch_size)

    loss_i = cross_entropy(similarity, labels)       # pull correct image->text pairs together
    loss_t = cross_entropy(similarity.T, labels)      # pull correct text->image pairs together
    loss = (loss_i + loss_t) / 2

    loss.backward()
    optimizer.step()

query_vector = embed("What's our refund policy?")
top_chunks = vector_db.search(query_vector, k=4)
context = "\n\n".join(chunk.text for chunk in top_chunks)
prompt = f"Answer using only this context:\n{context}\n\nQuestion: {query}"
answer = llm.generate(prompt)

import numpy as np

def cosine_similarity(a, b):
    a, b = np.array(a), np.array(b)
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

# Two sentences with related meaning should land close in vector space
vec_dog = embed_model.encode("the dog ran across the yard")
vec_puppy = embed_model.encode("a puppy sprinted through the grass")
vec_stock = embed_model.encode("quarterly earnings exceeded forecasts")

print(cosine_similarity(vec_dog, vec_puppy))  # high, close in meaning
print(cosine_similarity(vec_dog, vec_stock))  # low, unrelated topics

A moody cyberpunk alley at night, neon signs reflecting on wet asphalt,
a lone figure in a translucent raincoat walking away from camera,
volumetric fog, cinematic lighting, shot on 35mm film, shallow depth of field
--ar 16:9 --style raw

User audio -> streaming ASR -> partial transcript
                                        |
                                        v
                              LLM starts generating
                              on partial/final text
                                        |
                                        v
                         streaming TTS synthesizes
                         audio as tokens arrive
                                        |
                                        v
                              audio playback begins
                         before the full reply exists

Each arrow is a latency contributor. In text chat, only
the LLM step is visible to the user. In voice, all four
stages sit on the critical path between 'user stops
talking' and 'user hears a response.'

# Symmetric per-channel quantization of a weight tensor
# scale is computed independently for each output channel
import numpy as np

def quantize_per_channel(weights, num_bits=8):
    qmax = 2 ** (num_bits - 1) - 1  # 127 for int8
    scales = np.max(np.abs(weights), axis=1, keepdims=True) / qmax
    scales[scales == 0] = 1e-8  # avoid divide by zero on dead channels
    q_weights = np.round(weights / scales).clip(-qmax - 1, qmax)
    return q_weights.astype(np.int8), scales

def dequantize(q_weights, scales):
    return q_weights.astype(np.float32) * scales

function payAgent(address agent, uint task) external {
  require(oracle.verify(task), "unverified work");
  token.transfer(agent, bounty[task]);
}

# Open-loop: trust the plan, never check reality
for torque_command in precomputed_torque_sequence:
    send_to_motor(torque_command)
    wait(dt)
# If friction, an unexpected bump, or a bad mass estimate
# throws things off, nothing here will ever notice or correct it.

# Closed-loop: sense, compare, correct, repeat
while not at_target:
    actual_state = read_sensors()          # where is the joint really?
    error = desired_state - actual_state   # how far off are we?
    correction = compute_correction(error) # controller decides the fix
    torque_command = feedforward_torque + correction
    send_to_motor(torque_command)
    wait(dt)

# Simplified PID control loop, runs many times per second
integral = 0
previous_error = 0

def pid_step(setpoint, measured_value, dt, kp, ki, kd):
    global integral, previous_error

    error = setpoint - measured_value

    P = kp * error

    integral += error * dt
    I = ki * integral

    derivative = (error - previous_error) / dt
    D = kd * derivative

    previous_error = error

    correction = P + I + D
    return correction

# Simplified core loop of a sampling-based planner (RRT-style)
tree = Tree(root=start_config)

while not tree.contains(goal_config):
    random_config = sample_random_config(space)
    nearest = tree.nearest_neighbor(random_config)
    new_config = step_towards(nearest, random_config, max_step)

    if is_collision_free(nearest, new_config):
        tree.add_edge(nearest, new_config)

    if close_enough(new_config, goal_config):
        break

path = tree.trace_path_to(goal_config)

loop at high frequency (e.g. every 1-4 ms):
    state = read_sensors()               # IMU, joint encoders, foot contact
    com_xy = estimate_com_projection(state)
    support_polygon = compute_support_polygon(state.contact_feet)

    if com_xy inside support_polygon:
        margin = distance_to_nearest_edge(com_xy, support_polygon)
        if margin > safety_threshold:
            hold_current_gait_trajectory()
        else:
            nudge_next_footstep(toward = center_of(support_polygon))
    else:
        # dynamically unstable moment, must catch it, not just log it
        correction = compute_capture_step(state.com_velocity, support_polygon)
        command_swing_leg(correction)      # place foot to rebuild support under COM

    apply_joint_torques(state, target = corrected_trajectory)

joint_angles = inverse_kinematics(target_xyz=(0.4, 0.1, 0.3))
for angle, limit in zip(joint_angles, joint_limits):
    if abs(angle) > limit:
        raise MotionError('Target unreachable within joint limits')
arm.move_to(joint_angles)

# Simplified particle filter localization loop
  def particle_filter_step(particles, weights, motion_cmd, sensor_scan, known_map):
      # 1. PREDICT: move every particle by the commanded motion,
      #    plus a little random noise, since motion is never exact
      for p in particles:
          p.pose = apply_motion(p.pose, motion_cmd, noise=motion_noise())
  
      # 2. UPDATE: score each particle by how well its predicted
      #    sensor reading matches what the real sensor just saw
      for i, p in enumerate(particles):
          expected_scan = simulate_sensor(p.pose, known_map)
          weights[i] = measurement_likelihood(sensor_scan, expected_scan)
  
      weights = normalize(weights)
  
      # 3. RESAMPLE: keep particles proportional to their weight,
      #    good matches survive and multiply, bad ones die out
      particles = resample(particles, weights)
  
      belief_estimate = weighted_mean(particles, weights)
      return particles, belief_estimate

# Simplified dynamic window approach: score candidate velocities
  def dynamic_window_approach(robot, obstacles, goal, dt=0.1, horizon=1.5):
      best_score = -float("inf")
      best_v, best_w = 0.0, 0.0
  
      # Only consider velocities reachable given current speed + accel limits
      for v in reachable_linear_velocities(robot):
          for w in reachable_angular_velocities(robot):
              path = simulate_forward(robot, v, w, dt, horizon)
  
              if collides(path, obstacles):
                  continue  # not physically safe, discard immediately
  
              clearance = min_distance_to_obstacles(path, obstacles)
              progress = heading_alignment(path, goal)
              speed = v  # mild preference for not crawling
  
              score = 0.5 * progress + 0.3 * clearance + 0.2 * speed
  
              if score > best_score:
                  best_score = score
                  best_v, best_w = v, w
  
      return best_v, best_w  # re-run this whole search again next tick

# Simplified flocking rule for one robot in a swarm
  # Runs locally on every robot, using only nearby neighbors
  
  def compute_steering(self, neighbors):
      if not neighbors:
          return self.current_heading
  
      # Rule 1: move toward the average position of nearby neighbors
      avg_position = average([n.position for n in neighbors])
      cohesion_vector = avg_position - self.position
  
      # Rule 2: steer away from anything too close
      separation_vector = vector(0, 0)
      for n in neighbors:
          distance = dist(self.position, n.position)
          if distance < MIN_SAFE_DISTANCE:
              separation_vector += (self.position - n.position) / distance
  
      # Combine both rules, no global plan or central coordinator involved
      return normalize(cohesion_vector * 0.5 + separation_vector * 1.5)

policy = initialize_random_policy()
  
  for episode in range(num_episodes):
      state = env.reset()
      done = False
  
      while not done:
          action = policy(state)              # choose action from current policy
          next_state, reward, done = env.step(action)  # act in the environment
  
          buffer.store(state, action, reward, next_state)
          state = next_state
  
      policy = update_policy(policy, buffer)  # nudge policy toward higher-reward actions
  
  # after enough episodes, policy has learned a control strategy
  # purely from experience, with no hand-written control law

distance_cm = ultrasonic.read()
if distance_cm < 15:
    robot.stop()
else:
    robot.move_forward()

belief = init_particle_filter(n_particles=500)
while robot.running():
    belief = predict(belief, robot.odometry())
    belief = update(belief, robot.sensor_scan(), occupancy_map)
    pose_estimate = belief.weighted_mean()
    occupancy_map = integrate_scan(occupancy_map, robot.sensor_scan(), pose_estimate)

PERCEPTION-TO-GRASP PIPELINE

  camera / depth sensor
        |
        v
  segmentation  ->  isolate target object from clutter
        |
        v
  pose estimation  ->  where is it, which way is it facing
        |
        v
  grasp candidate generation  ->  many possible grip points
        |
        v
  grasp scoring  ->  rank by force closure, stability, reachability
        |
        v
  execute + verify  ->  did we actually get it, retry if not

# Behavioral cloning: supervised learning over (state, action) pairs
  # collected from expert demonstrations
  
  import torch.nn as nn
  
  policy = nn.Sequential(
      nn.Linear(state_dim, 256), nn.ReLU(),
      nn.Linear(256, 256), nn.ReLU(),
      nn.Linear(256, action_dim),
  )
  
  # states, actions come from logged expert trajectories,
  # not from the policy's own rollouts
  for states, actions in expert_dataloader:
      predicted = policy(states)
      loss = mse_loss(predicted, actions)
      loss.backward()
      optimizer.step()
  
  # DAgger's key change: periodically roll out policy itself,
  # ask the expert to label the states it visited, and add those
  # labeled states back into the training set before the next round

# Simplified VLA inference loop running on a humanoid robot
while task_active:
    image = camera.capture()          # current visual observation
    state = robot.proprioception()    # joint angles, velocities, torques

    # single forward pass: perception + language + action in one model
    action_chunk = vla_model.predict(
        image=image,
        instruction="pick up the red mug and place it on the shelf",
        proprio=state,
    )

    for action in action_chunk:       # execute a short horizon of actions
        robot.apply_joint_targets(action)
        if safety_monitor.violation_detected():
            robot.freeze()
            break

# Simplified VLA inference loop running on a humanoid robot
while task_active:
    image = camera.capture()          # current visual observation
    state = robot.proprioception()    # joint angles, velocities, torques

    # single forward pass: perception + language + action in one model
    action_chunk = vla_model.predict(
        image=image,
        instruction="pick up the red mug and place it on the shelf",
        proprio=state,
    )

    for action in action_chunk:       # execute a short horizon of actions
        robot.apply_joint_targets(action)
        if safety_monitor.violation_detected():
            robot.freeze()
            break

# Simplified motion-scaling and tremor-filter pipeline
  # between a surgeon's console input and an instrument tip
  
  def process_input(raw_position, prev_filtered, scale_factor=0.2, alpha=0.15):
      # Low-pass filter to attenuate high-frequency hand tremor
      # (tremor is fast; intentional motion is comparatively slow)
      filtered = alpha * raw_position + (1 - alpha) * prev_filtered
  
      # Motion scaling: large hand movement at the console becomes
      # a small, precise movement at the instrument tip
      scaled_delta = (filtered - prev_filtered) * scale_factor
  
      target_position = prev_filtered_output + scaled_delta
      return target_position, filtered

// Minimal obstacle-avoidance control loop for a differential-drive rover
  const int TRIG_PIN = 9, ECHO_PIN = 10;
  const int LEFT_PWM = 5, LEFT_DIR = 4, RIGHT_PWM = 6, RIGHT_DIR = 7;
  const int STOP_DISTANCE_CM = 20;
  
  long readDistanceCm() {
    digitalWrite(TRIG_PIN, LOW); delayMicroseconds(2);
    digitalWrite(TRIG_PIN, HIGH); delayMicroseconds(10);
    digitalWrite(TRIG_PIN, LOW);
    long duration = pulseIn(ECHO_PIN, HIGH, 30000); // 30ms timeout
    if (duration == 0) return 999; // no echo, treat as clear
    return duration / 58;
  }
  
  void driveMotors(int leftSpeed, int rightSpeed) {
    digitalWrite(LEFT_DIR, leftSpeed >= 0 ? HIGH : LOW);
    digitalWrite(RIGHT_DIR, rightSpeed >= 0 ? HIGH : LOW);
    analogWrite(LEFT_PWM, abs(leftSpeed));
    analogWrite(RIGHT_PWM, abs(rightSpeed));
  }
  
  void loop() {
    long distance = readDistanceCm();
    if (distance < STOP_DISTANCE_CM) {
      driveMotors(-120, 120); // pivot in place to find clear space
      delay(300);
    } else {
      driveMotors(150, 150); // drive forward
    }
    delay(50); // loop pacing, not a real-time guarantee
  }

def image_callback(msg):
    detections = detect_objects(msg)
    detection_pub.publish(detections)

rospy.Subscriber('/camera/image', Image, image_callback)
detection_pub = rospy.Publisher('/detections', Detections, queue_size=10)

# Simplified closed-loop velocity control for a geared DC motor
  def velocity_control_step(target_rpm, encoder_ticks, dt, gear_ratio, ticks_per_rev):
      measured_rpm = (encoder_ticks / ticks_per_rev) * (60 / dt) / gear_ratio
      error = target_rpm - measured_rpm
      integral_error += error * dt
      duty_cycle = clamp(Kp * error + Ki * integral_error, 0.0, 1.0)
      pwm_write(motor_channel, duty_cycle)
      return measured_rpm, duty_cycle

// Bare-metal-style motor control loop on an MCU, fixed-rate, deterministic
  void control_loop_1kHz(void) {
      // Called from a hardware timer interrupt, exactly every 1ms
      int16_t target_rpm = read_latest_command();      // from SBC over UART/CAN
      int16_t actual_rpm = read_encoder_delta();        // from quadrature encoder
      int16_t error = target_rpm - actual_rpm;
  
      pid_state.integral += error;
      int16_t output = (Kp * error) + (Ki * pid_state.integral) - (Kd * pid_state.last_error);
      pid_state.last_error = error;
  
      if (limit_switch_triggered() || overcurrent_detected()) {
          set_pwm_duty(0);          // safety interlock overrides everything
      } else {
          set_pwm_duty(clamp(output, -MAX_DUTY, MAX_DUTY));
      }
  }

# Rough power budget check before committing to a battery pack
  def runtime_estimate(capacity_mAh, pack_voltage, loads_watts, safety_margin=0.8):
      total_watts = sum(loads_watts.values())
      pack_wh = (capacity_mAh / 1000) * pack_voltage
      usable_wh = pack_wh * safety_margin  # never plan to fully drain a LiPo
      hours = usable_wh / total_watts
      return hours * 60  # minutes of usable runtime
  
  loads = {"drive_motors": 45.0, "compute": 9.0, "sensors": 4.5, "servos": 6.0}
  print(runtime_estimate(4000, 11.1, loads))  # 3S 4000mAh pack, minutes

Transaction inputs:
  1.2 BTC (from a previous payment you received)

Transaction outputs:
  0.5 BTC -> recipient's address
  0.68 BTC -> change, back to you (minus a small fee)

The 1.2 BTC input is fully "spent" and can never be
reused. It splits into two new outputs, one of which
becomes a new spendable unit (a UTXO) sitting in your
wallet until you spend it again.

# A standard Pay-to-Pubkey-Hash (P2PKH) locking script
OP_DUP OP_HASH160 <pubKeyHash> OP_EQUALVERIFY OP_CHECKSIG

# To spend it, the unlocking script supplies:
<signature> <publicKey>

# The two scripts run together on Bitcoin's stack machine:
# 1. Push signature and public key onto the stack
# 2. Duplicate the public key, hash it, compare to pubKeyHash
# 3. If it matches, verify the signature against the public key
# No loops, no external calls, just a fixed sequence of stack ops.

use bitcoin::{Amount, Transaction, TxIn, TxOut, Sequence, Witness, ScriptBuf};
use bitcoin::transaction::Version;
use bitcoin::locktime::absolute::LockTime;
use bitcoin::{OutPoint, Txid};
use std::str::FromStr;

fn build_transaction() -> Transaction {
    let prev_txid =
        Txid::from_str("f4184fc596403b9d638783cf57adfe4c75c605f6356fbc91338530e9831e9e16")
            .expect("valid txid");

    let input = TxIn {
        previous_output: OutPoint::new(prev_txid, 0),
        script_sig: ScriptBuf::new(),
        sequence: Sequence::ENABLE_RBF_NO_LOCKTIME,
        witness: Witness::new(),
    };

    let recipient_script = ScriptBuf::new_p2wpkh(&"a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2"
        .parse()
        .expect("valid pubkey hash"));

    let output = TxOut {
        value: Amount::from_sat(50_000),
        script_pubkey: recipient_script,
    };

    Transaction {
        version: Version::TWO,
        lock_time: LockTime::ZERO,
        input: vec![input],
        output: vec![output],
    }
}

fn main() {
    let tx = build_transaction();
    println!("txid: {}", tx.compute_txid());
    println!("weight: {} wu", tx.weight());
}

// A minimal smart contract: stores a number, anyone can read it,
// only the owner who deployed it can change it.
contract SimpleStorage {
    uint256 public storedNumber;
    address public owner;

    constructor() {
        owner = msg.sender;
    }

    function setNumber(uint256 newNumber) public {
        require(msg.sender == owner, "Not the owner");
        storedNumber = newNumber;
    }
}

// Vulnerable withdraw pattern vs. the checks-effects-interactions fix
contract VaultVulnerable {
    mapping(address => uint256) public balances;

    function deposit() external payable {
        balances[msg.sender] += msg.value;
    }

    // BAD: sends funds before updating the balance, opens a reentrancy hole
    function withdraw(uint256 amount) external {
        require(balances[msg.sender] >= amount, "insufficient balance");
        (bool sent, ) = msg.sender.call{value: amount}("");
        require(sent, "transfer failed");
        balances[msg.sender] -= amount; // too late, attacker already re-entered
    }
}

contract VaultFixed {
    mapping(address => uint256) public balances;

    function deposit() external payable {
        balances[msg.sender] += msg.value;
    }

    // GOOD: checks-effects-interactions, state is updated before the external call
    function withdraw(uint256 amount) external {
        require(balances[msg.sender] >= amount, "insufficient balance");
        balances[msg.sender] -= amount;
        (bool sent, ) = msg.sender.call{value: amount}("");
        require(sent, "transfer failed");
    }
}

struct UserOperation {
    address sender;          // the smart contract wallet
    uint256 nonce;
    bytes   initCode;        // deploys wallet if it doesn't exist yet
    bytes   callData;        // what the wallet should do
    uint256 callGasLimit;
    uint256 verificationGasLimit;
    uint256 preVerificationGas;
    uint256 maxFeePerGas;
    uint256 maxPriorityFeePerGas;
    bytes   paymasterAndData; // who pays, and how
    bytes   signature;
}

// A bundler collects UserOperations and calls this on the EntryPoint contract
function handleOps(UserOperation[] calldata ops, address payable beneficiary) external;

#[derive(Accounts)]
  pub struct UpdateCounter<'info> {
      #[account(mut)]
      pub counter: Account<'info, Counter>,
      pub authority: Signer<'info>,
  }
  
  pub fn increment(ctx: Context<UpdateCounter>) -> Result<()> {
      // The program itself holds no state.
      // It only ever operates on the accounts handed to it here.
      ctx.accounts.counter.count += 1;
      Ok(())
  }

l1.postBatch(rollup.compress(1000_txs))
// 1000 transactions, 1 L1 fee

// Simplified lock-and-mint bridge contract (source chain side)
  contract LockAndMintBridge {
      mapping(bytes32 => bool) public processedWithdrawals;
      address public validatorSet; // multisig or light client verifier
  
      event Locked(address indexed user, uint256 amount, uint256 destChainId);
  
      function lock(uint256 amount, uint256 destChainId) external {
          token.transferFrom(msg.sender, address(this), amount);
          emit Locked(msg.sender, amount, destChainId);
          // Off-chain relayers/validators observe this event and
          // sign an attestation authorizing a mint on the destination chain
      }
  
      function release(bytes32 withdrawalId, address to, uint256 amount, bytes calldata proof) external {
          require(!processedWithdrawals[withdrawalId], "already processed");
          require(verify(proof), "invalid proof or signatures"); // the entire security model lives here
          processedWithdrawals[withdrawalId] = true;
          token.transfer(to, amount);
      }
  }

// Simplified over-collateralized vault: mint stablecoin against locked collateral
  function mint(uint256 collateralAmount, uint256 debtAmount) external {
      require(collateralAmount > 0, "no collateral");
      uint256 collateralValue = oracle.getPrice(collateralToken) * collateralAmount;
  
      // Must stay above the minimum collateralization ratio, e.g. 150%
      require(
          collateralValue * 100 >= debtAmount * MIN_COLLATERAL_RATIO,
          "below min collateral ratio"
      );
  
      vaults[msg.sender].collateral += collateralAmount;
      vaults[msg.sender].debt += debtAmount;
  
      collateralToken.transferFrom(msg.sender, address(this), collateralAmount);
      stablecoin.mint(msg.sender, debtAmount);
  }