feat(java): proxy sessions with identity dedup and cleanup

sdks/java/src/main/java/org/byteveda/taskito/proxies/Proxies.java

@@ -84,12 +84,32 @@ public Object reconstruct(ProxyRef ref) {
      * Verify a ref's signature, expiry, and (when {@code expectedPurpose} is
      * non-null) its bound purpose, then reconstruct the resource.
      */
-    @SuppressWarnings("unchecked")
     public Object reconstruct(ProxyRef ref, String expectedPurpose) {
-        ProxyHandler<Object> handler = (ProxyHandler<Object>) handlers.get(ref.handler());
+        ProxyHandler<Object> handler = handlerFor(ref.handler());
+        verify(ref, expectedPurpose);
+        return handler.reconstruct(ref.reference());
+    }
+
+    /**
+     * A new {@link ProxySession}: deconstruct/reconstruct with per-instance
+     * identity dedup and close-time cleanup. Confine a session to one thread.
+     */
+    public ProxySession session() {
+        return new ProxySession(this);
+    }
+
+    /** Look up a handler by id, failing on an unknown id. */
+    @SuppressWarnings("unchecked")
+    ProxyHandler<Object> handlerFor(String handlerId) {
+        ProxyHandler<Object> handler = (ProxyHandler<Object>) handlers.get(handlerId);
         if (handler == null) {
-            throw new ProxyException("unknown proxy handler '" + ref.handler() + "'");
+            throw new ProxyException("unknown proxy handler '" + handlerId + "'");
         }
+        return handler;
+    }
+
+    /** Verify a ref's signature, expiry, and (when requested) bound purpose. */
+    void verify(ProxyRef ref, String expectedPurpose) {
         byte[] expected = sign(ref.handler(), ref.reference(), ref.expiresAtMs(), ref.purpose())
                 .getBytes(StandardCharsets.UTF_8);
         byte[] actual = (ref.signature() == null ? "" : ref.signature()).getBytes(StandardCharsets.UTF_8);
@@ -102,7 +122,6 @@ public Object reconstruct(ProxyRef ref, String expectedPurpose) {
         if (expectedPurpose != null && !expectedPurpose.equals(ref.purpose())) {
             throw new ProxyException("proxy purpose mismatch for handler '" + ref.handler() + "'");
         }
-        return handler.reconstruct(ref.reference());
     }
 
     /** {@link #reconstruct(ProxyRef)} cast to the caller's type. */

sdks/java/src/main/java/org/byteveda/taskito/proxies/ProxyHandler.java

@@ -21,4 +21,12 @@ public interface ProxyHandler<T> {
 
     /** Rebuild the resource from a reference produced by {@link #deconstruct}. */
     T reconstruct(Map<String, Object> reference);
+
+    /**
+     * Release a resource this handler reconstructed, invoked (LIFO, once per
+     * unique instance) when a {@link ProxySession} that produced it closes.
+     * Direct {@link Proxies#reconstruct(ProxyRef)} calls have no lifecycle and
+     * never trigger this. Default: no-op.
+     */
+    default void cleanup(T value) {}
 }

sdks/java/src/main/java/org/byteveda/taskito/proxies/ProxySession.java

@@ -0,0 +1,151 @@
+package org.byteveda.taskito.proxies;
+
+import java.lang.System.Logger;
+import java.lang.System.Logger.Level;
+import java.time.Duration;
+import java.util.ArrayDeque;
+import java.util.Deque;
+import java.util.HashMap;
+import java.util.IdentityHashMap;
+import java.util.Map;
+import org.byteveda.taskito.errors.ProxyException;
+
+/**
+ * A unit-of-work wrapper over {@link Proxies} adding identity dedup and a
+ * cleanup lifecycle. Within one session:
+ *
+ * <ul>
+ *   <li>{@link #deconstruct(Object)} memoizes by (instance identity, purpose) —
+ *       deconstructing the same object again returns the same {@link ProxyRef}
+ *       without calling the handler twice. The first call's TTL wins per
+ *       (instance, purpose); use a new session to refresh expiry.
+ *   <li>{@link #reconstruct(ProxyRef)} memoizes by the ref's signature — every
+ *       ref to the same underlying resource resolves to the same instance, and
+ *       the handler reconstructs it once. Signature, expiry, and purpose are
+ *       re-verified on every call, memo hit or not.
+ *   <li>{@link #close()} runs {@link ProxyHandler#cleanup} once per unique
+ *       reconstructed instance, in reverse reconstruction order (LIFO).
+ * </ul>
+ *
+ * <p>Not thread-safe — confine a session to the thread that created it. A
+ * session models one producer batch or one task invocation.
+ */
+public final class ProxySession implements AutoCloseable {
+    private static final Logger LOG = System.getLogger(ProxySession.class.getName());
+
+    private final Proxies proxies;
+    private final IdentityHashMap<Object, Map<String, ProxyRef>> deconstructed = new IdentityHashMap<>();
+    private final Map<String, Object> reconstructed = new HashMap<>();
+    private final Deque<Runnable> cleanups = new ArrayDeque<>();
+    private boolean closed;
+
+    ProxySession(Proxies proxies) {
+        this.proxies = proxies;
+    }
+
+    /** Deconstruct {@code value} (no expiry or purpose), deduped by instance identity. */
+    public ProxyRef deconstruct(Object value) {
+        return deconstruct(value, null, null);
+    }
+
+    /** Deconstruct {@code value} with a TTL, deduped by instance identity. */
+    public ProxyRef deconstruct(Object value, Duration ttl) {
+        return deconstruct(value, ttl, null);
+    }
+
+    /**
+     * Deconstruct {@code value} bound to {@code ttl}/{@code purpose} (both
+     * nullable), deduped by (instance identity, purpose).
+     */
+    public ProxyRef deconstruct(Object value, Duration ttl, String purpose) {
+        ensureOpen();
+        if (value == null) {
+            throw new ProxyException("cannot deconstruct null");
+        }
+        Map<String, ProxyRef> byPurpose = deconstructed.computeIfAbsent(value, key -> new HashMap<>());
+        ProxyRef cached = byPurpose.get(purpose);
+        if (cached != null) {
+            return cached;
+        }
+        ProxyRef ref = proxies.deconstruct(value, ttl, purpose);
+        byPurpose.put(purpose, ref);
+        return ref;
+    }
+
+    /** Verify and reconstruct {@code ref}, deduped by its signature. */
+    public Object reconstruct(ProxyRef ref) {
+        return reconstruct(ref, null);
+    }
+
+    /**
+     * Verify (always — including memo hits, so a ref that expires mid-session
+     * stops resolving) and reconstruct {@code ref}, deduped by its signature.
+     */
+    public Object reconstruct(ProxyRef ref, String expectedPurpose) {
+        ensureOpen();
+        ProxyHandler<Object> handler = proxies.handlerFor(ref.handler());
+        proxies.verify(ref, expectedPurpose);
+        String signature = ref.signature();
+        if (reconstructed.containsKey(signature)) {
+            return reconstructed.get(signature);
+        }
+        Object value = handler.reconstruct(ref.reference());
+        reconstructed.put(signature, value);
+        cleanups.push(() -> handler.cleanup(value));
+        return value;
+    }
+
+    /** {@link #reconstruct(ProxyRef)} cast to the caller's type. */
+    @SuppressWarnings("unchecked")
+    public <T> T resolve(ProxyRef ref) {
+        return (T) reconstruct(ref);
+    }
+
+    /** {@link #reconstruct(ProxyRef, String)} cast to the caller's type. */
+    @SuppressWarnings("unchecked")
+    public <T> T resolve(ProxyRef ref, String expectedPurpose) {
+        return (T) reconstruct(ref, expectedPurpose);
+    }
+
+    /**
+     * Run {@link ProxyHandler#cleanup} for every reconstructed instance in
+     * reverse order (LIFO), once each. A cleanup failure is logged and never
+     * skips the rest; a JVM-level {@link Error} is rethrown only after the
+     * remaining cleanups have drained. Idempotent.
+     */
+    @Override
+    public void close() {
+        if (closed) {
+            return;
+        }
+        closed = true;
+        Error fatal = null;
+        while (!cleanups.isEmpty()) {
+            Runnable cleanup = cleanups.pop();
+            try {
+                cleanup.run();
+            } catch (RuntimeException e) {
+                // Cleanup must never fail the rest of the teardown — record and continue.
+                LOG.log(Level.WARNING, "proxy cleanup failed", e);
+            } catch (Error e) {
+                // Drain the remaining cleanups first, then let the error propagate —
+                // aborting here would abandon them for good (closed is already set).
+                if (fatal == null) {
+                    fatal = e;
+                }
+                LOG.log(Level.WARNING, "proxy cleanup threw an error", e);
+            }
+        }
+        deconstructed.clear();
+        reconstructed.clear();
+        if (fatal != null) {
+            throw fatal;
+        }
+    }
+
+    private void ensureOpen() {
+        if (closed) {
+            throw new ProxyException("proxy session is closed");
+        }
+    }
+}