Listing: Listing

def proposed_tests():
    # These are specifications; no executed pass is asserted.
    points = torch.tensor(
        [[2., 1.], [1., 2.], [-2., 1.], [-1., 2.],
         [-2., -1.], [-1., -2.], [2., -1.], [1., -2.]],
        dtype=torch.float64
    )
    table = rule_table(points)
    assert torch.equal(table[:, 1], table[:, 2])
    votes = torch.stack(
        [(table == label).sum(dim=1) for label in range(4)], dim=1
    )
    assert float(votes.max(dim=1).values.double().mean()) / 4.0 == 0.75
    chosen_votes = (table == table[:, 1, None]).sum(dim=1)
    assert torch.equal(chosen_votes, votes.max(dim=1).values)

    expected = torch.tensor(
        [[0., .5, .5, .75], [.5, 0., 0., .5],
         [.5, 0., 0., .5], [.75, .5, .5, 0.]],
        dtype=torch.float64
    )
    observed = torch.empty((4, 4), dtype=torch.float64)
    for r in range(4):
        for s in range(4):
            observed[r, s] = (table[:, r] != table[:, s]).double().mean()
    assert torch.equal(observed, expected)

    # Old-function responses without adjacent response repetition.
    alternating = torch.tensor(
        [[-2., 1.], [2., -1.]] * 6, dtype=torch.float64
    )
    rules = torch.tensor([1] + [0] * 11, dtype=torch.long)
    old_predictions = rule_table(alternating)[:, 1]
    metrics = sequence_metrics(
        alternating, rules, old_predictions, post_window=10
    )
    assert metrics["old_fraction_sample"] == 1.0
    assert metrics["repeated_error_fraction_sample"] == 0.0

    constant = torch.zeros(12, dtype=torch.long)
    metrics = sequence_metrics(alternating, rules, constant, post_window=10)
    assert metrics["old_fraction_sample"] == 0.0
    assert metrics["repeated_error_fraction_sample"] == 1.0

    # Zero-denominator certification remains unresolved.
    certificate = ratio_certificate(
        [{"p": 0.0, "n": 0.0}] * 20, "p", "n"
    )
    assert certificate["status"] == "unresolved_denominator"

    event = first_completion(torch.tensor([1, 1, 1, 1, 0]))
    censored = first_completion(torch.tensor([0, 0, 0, 0, 0]))
    assert event["first_completion"] == 5 and not event["censored"]
    assert censored["first_completion"] is None and censored["censored"]
    assert event["restricted_time"] == censored["restricted_time"]

    layer = ReferenceMaskedLinear(2, 1)
    with torch.no_grad():
        layer.weight.copy_(torch.tensor([[3., 2.]], dtype=torch.float64))
    layer.capture_dense_reference()
    deleted = layer.delete_exact(1)
    assert deleted.tolist() == [1]

    layer.zero_grad(set_to_none=True)
    layer.begin_scoring()
    loss = layer(torch.ones((1, 2), dtype=torch.float64)).square().sum()
    loss.backward()
    effective_score = layer.effective_gradient().clone()
    assert layer.weight.grad[0, 1].item() == 0.0
    assert effective_score[0, 1].item() != 0.0
    layer.end_scoring()

    clone = copy.deepcopy(layer)
    result = clone.reopen_exact(1, effective_score, scale=1.0)
    assert result["selected"].tolist() == [1]
    assert clone.weight[0, 1].item() == 2.0
    assert layer.weight[0, 1].item() == 0.0

    # Exact interpolation remainder for q(d)=d^2 on [0,1].
    rows = [
        {"candidate_id": "left", "displacement": 0.0,
         "endpoint": 0.0, "radius": 0.0, "path_id": "quadratic"},
        {"candidate_id": "right", "displacement": 1.0,
         "endpoint": 1.0, "radius": 0.0, "path_id": "quadratic"},
    ]
    record = matching_record(rows, 0.5, 0.01, curvature=2.0)
    assert record["status"] == "bracketed_surrogate"
    assert record["estimate"] == 0.5
    assert record["allowance"] == 0.25
    assert abs(record["estimate"] - 0.5**2) == record["allowance"]

    # Feedback inputs contain the preceding label only.
    task = make_uncued_task(3, 12, 4, seed=0)
    feedback = observation_tensor(task, "past_target_feedback")
    assert bool((feedback[:, 0, 2:] == 0).all())
    assert torch.equal(
        feedback[:, 1:, 2:].argmax(dim=-1),
        task["labels"][:, :-1]
    )
    assert observation_tensor(task, "uncued").shape[-1] == 2